Vector Retrieval Service for Milvus:Alibaba Cloud Milvus ナレッジベースを使用したインテリジェントなカスタマーサービス Q&A アプリケーションの構築

ソリューション概要

全体的なパイプラインは次のとおりです：コンソールでタグを定義してナレッジベースを作成 → タグごとにカスタマーサービスドキュメントをバッチインポート → バージョンを公開 → SDK を介して検索し、必要に応じてタグでフィルタリング → 結果を LLM に渡して出典付きの回答を生成 → Flask で Q&A ページを提供。

このトピックでは、カスタマーサービスシナリオに特有の 3 つの要素、タグ (メタデータ) フィルタリング、リランクモデル、および 回答のトレーサビリティ に焦点を当てています。基本的なナレッジベースパイプライン (署名付きアップロード、データ登録、バージョンの公開) と最もシンプルな Q&A の実装については、個人ナレッジ Q&A アプリケーションの構築をご参照ください。

開始する前に、PDF、DOCX、Markdown、または TXT フォーマットのドキュメントを 5 ～ 20 件準備してください。全体のプロセスには約 20 ～ 30 分かかります。ドキュメント解析時間は、ドキュメントの数とサイズによって異なります。

前提条件

• ナレッジベースをサポートするリージョン (China (Hangzhou)、China (Beijing)、China (Zhangjiakou)、または China (Shenzhen)) でナレッジベースを作成します。ナレッジベース ID を記録してください。形式は kd-803ae9b10cc31 です。

• Alibaba Cloud アカウントから RAM ユーザーを作成し、[永続的な AccessKey を使用] を選択して、システムポリシー AliyunMilvusFullAccess を RAM ユーザーにアタッチします。

• OpenAI の chat/completions プロトコルをサポートする LLM エンドポイントと API キーを用意します。例えば、Alibaba Cloud Model Studio の OpenAI 互換エンドポイントを使用します。

• ローカルマシンに Python 3.8 以降をインストールしておきます。

  認証情報のセキュリティと本番デプロイのガイダンスについては、ローンチ前のチェックリスト をご参照ください。

ステップ 1：タグの定義

タグは、ドキュメントにビジネスディメンション (ドキュメントタイプ、製品ライン、有効日など) をアタッチします。タグはインポート時にナレッジベースに書き込まれ、検索時のフィルタリングに使用されます。カスタマーサービスシナリオにおいて、ポリシー、マニュアル、FAQ を区別するための重要な要素です。

1. Alibaba Cloud Milvus コンソール にログインし、目的のナレッジベースの詳細ページを開きます。

2. [Basic Information] で [Tags] を見つけ、[Manage] をクリックします。

3. [タグ管理] ダイアログボックスで、タグ名を入力し、フィールドタイプを選択して、[追加] をクリックします。このトピックでは、次の 3 つのタグを使用します。すべて string 型です。

   タグ名	説明	値の例
   docType	ドキュメントタイプ	policy, manual, faq
   productLine	適用される製品ライン	all, phone
   effectiveDate	有効日	2026-01-01

4. 3 つのタグを追加したら、[Done] をクリックします。詳細ページの [Tags] に「3 tags」と表示されます。

   本チュートリアルでは、ステップ 5 のインポート時にタグ値をドキュメントに一括で書き込むため、現時点でコンソールで値を設定する必要はありません。ナレッジベースに既に存在するドキュメントにタグを付ける場合は、FAQ に記載されているように、[Data Management] ページの [Set Tags] を使用します。

ステップ 2：ドキュメントとインポートマニフェストの準備

• カスタマーサービスのドキュメントをローカルの documents/ ディレクトリに配置します。例：

kb-demo/
├── documents/
│   ├── shipping-policy.md
│   ├── return-policy.md
│   ├── warranty-policy.md
│   ├── phone-manual.md
│   └── invoice-faq.md

• documents.jsonl を作成します。各行に 1 つのドキュメントとそのタグを記述します。タグ名は、ステップ 1 で定義したものと一致させる必要があります。

{"path": "documents/shipping-policy.md", "metadata": {"docType": "policy", "productLine": "all", "effectiveDate": "2026-01-01"}}
{"path": "documents/return-policy.md", "metadata": {"docType": "policy", "productLine": "all", "effectiveDate": "2026-01-01"}}
{"path": "documents/warranty-policy.md", "metadata": {"docType": "policy", "productLine": "phone", "effectiveDate": "2026-03-01"}}
{"path": "documents/phone-manual.md", "metadata": {"docType": "manual", "productLine": "phone", "effectiveDate": "2026-03-01"}}
{"path": "documents/invoice-faq.md", "metadata": {"docType": "faq", "productLine": "all", "effectiveDate": "2026-02-01"}}

ファイル名はドキュメントのトピックを反映させ、内容は完全なタイトルとコンテキストを含めるようにしてください。これにより、検索機能がステップ 3 で設定されたサンプル質問を含む実際の質問と一致するようになります。公式に有効なポリシーをインポートすることを優先し、相反する古いバージョンを同時に保持することは避けてください。

ステップ 3： ランタイムパラメーターの設定

• プロジェクトディレクトリを作成し、依存関係をインストールします。

mkdir -p kb-demo/documents kb-demo/templates
cd kb-demo
python3 -m venv .venv
source .venv/bin/activate
pip install Flask==3.1.1 requests==2.32.4 alibabacloud-milvusknowledgebase20260604==1.0.0

上記のパッケージは、kb_client.py によってインポートされる alibabacloud_tea_openapi や Flask がテンプレートのレンダリングに使用する Jinja2 などの追加の依存関係を自動的にインストールします。これらの依存関係を別途インストールする必要はありません。

• config.json を作成し、ご自身の認証情報とナレッジベース情報を入力します。

{
  "aliyun": {
    "access_key_id": "YOUR_ACCESS_KEY_ID",
    "access_key_secret": "YOUR_ACCESS_KEY_SECRET",
    "region_id": "cn-hangzhou",
    "knowledge_base_id": "kd-xxxxxxxxxxxxx",
    "knowledge_base_version": "LATEST_PUBLISHED"
  },
  "llm": {
    "enabled": true,
    "base_url": "https://YOUR_OPENAI_COMPATIBLE_ENDPOINT/v1",
    "api_key": "YOUR_LLM_API_KEY",
    "model": "YOUR_MODEL_NAME"
  },
  "upload": {
    "default_meta_fields": {}
  },
  "retrieval": {
    "page_size": 6,
    "candidate_count": 48,
    "min_score": 0.35,
    "semantic_weight": 0.7,
    "enable_query_expansion": true,
    "rerank_model_name": "qwen3-rerank",
    "tag_filter": {
      "relation": "and",
      "conditions": []
    }
  },
  "scenario": {
    "title": "Intelligent Customer Service Knowledge Base Q&A",
    "system_prompt": "あなたは企業のカスタマーサービスアシスタントです。取得されたドキュメントにのみ基づいて、明確でフレンドリーなカスタマーサービスの口調で回答してください。重要な結論には [ソース N] を引用してください。取得されたドキュメントがユーザーの質問を直接カバーしていない場合は、『利用可能なドキュメントでは、この内容について明確にカバーされていません』と回答する必要があります。部分的に関連するドキュメントから結論を推論したり、ドキュメントに記載されていない連絡先やサービスチャネルを追加したりしないでください。",
    "image_enabled": false,
    "sample_questions": [
      "支払い後、商品の発送まで通常どのくらいかかりますか？",
      "保証期間が終了した後でも、製品の修理は可能ですか？",
      "返品をリクエストするには、どのような条件を満たす必要がありますか？"
    ]
  }
}

このシナリオにおけるレトリーバルパラメーター値に関する考慮事項：

• min_score=0.35： 回答に含まれる、関連性の低いコンテンツを減らします。

• semantic_weight=0.7： セマンティック検索に重点を置き、口語的なカスタマーサービスの質問に対応します。

• rerank_model_name： 候補チャンクをリランクし、トップ結果の精度を向上させます。

• tag_filter.conditions： 空白のままにするとフィルタリングされません。タグフィルターについては、ステップ 7 をご参照ください。

ステップ 4：ナレッジベースクライアントの作成

以下の内容を kb_client.py として保存します。このコードは、タグ付きアップロードと、フィルタリングによる検索をカプセル化しています。

"""Milvus ナレッジベース OpenAPI SDK：タグ付きローカルアップロードと公開バージョンを対象とした検索。"""

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping, Sequence

import requests
from alibabacloud_milvusknowledgebase20260604 import models as models
from alibabacloud_milvusknowledgebase20260604.client import Client
from alibabacloud_tea_openapi import models as openapi_models

@dataclass(frozen=True)
class LocalDocument:
    path: Path
    object_path: str

    @classmethod
    def from_path(cls, value: str | Path) -> "LocalDocument":
        path = Path(value).expanduser().resolve()
        if not path.is_file():
            raise FileNotFoundError(path)
        return cls(path=path, object_path=path.name)

@dataclass(frozen=True)
class TagCondition:
    field: str
    op: str
    value: Any

@dataclass(frozen=True)
class SearchOptions:
    version: str = "LATEST_PUBLISHED"
    page_size: int = 6
    candidate_count: int = 48
    min_score: float = 0.0
    semantic_weight: float = 0.5
    enable_query_expansion: bool = True
    rerank_model_name: str | None = None
    tag_relation: str = "and"
    tag_conditions: tuple[TagCondition, ...] = ()

class KnowledgeBaseClient:
    def __init__(
        self,
        access_key_id: str,
        access_key_secret: str,
        region_id: str,
    ) -> None:
        endpoint = f"milvusknowledgebase.{region_id}.aliyuncs.com"
        self.client = Client(
            openapi_models.Config(
                access_key_id=access_key_id,
                access_key_secret=access_key_secret,
                region_id=region_id,
                endpoint=endpoint,
                connect_timeout=10_000,
                read_timeout=60_000,
            )
        )
        self.http = requests.Session()

    @staticmethod
    def _check(body: Any, action: str) -> None:
        # 成功したレスポンスの code は None、0、または "0" です。それ以外の明示的な code は失敗として扱います。
        if body is None:
            raise RuntimeError(f"{action} が空のレスポンスを返しました")
        if getattr(body, "success", None) is False or (
            getattr(body, "code", None) not in (None, 0, "0")
        ):
            raise RuntimeError(
                f"{action} が失敗しました: {getattr(body, 'message', 'unknown')}; "
                f"requestId={getattr(body, 'request_id', '')}"
            )

    def upload(
        self,
        knowledge_base_id: str,
        file_paths: Sequence[str | Path],
        meta_fields: Mapping[str, Any] | None = None,
    ) -> dict[str, Any]:
        docs = [LocalDocument.from_path(path) for path in file_paths]
        presign_docs = [
            models.GetKnowledgeBasePreSignedUrlRequestDocuments(
                path=doc.object_path,
                name=doc.path.name,
                size=doc.path.stat().st_size,
            )
            for doc in docs
        ]
        response = self.client.get_knowledge_base_pre_signed_url(
            knowledge_base_id,
            models.GetKnowledgeBasePreSignedUrlRequest(
                knowledge_base_id=knowledge_base_id,
                documents=presign_docs,
                expires_in=3600,
            ),
        )
        body = response.body
        self._check(body, "GetKnowledgeBasePreSignedUrl")

        urls = list(body.data.pre_signed_urls or [])
        if len(urls) != len(docs):
            raise RuntimeError("署名付き URL の数がファイルの数と一致しません")

        for doc, url in zip(docs, urls, strict=True):
            with doc.path.open("rb") as source:
                # 署名付き URL は空の Content-Type で署名されています。リクエストに Content-Type を含めないでください。
                self.http.put(url, data=source, timeout=120).raise_for_status()

        add_docs = [
            models.AddDocumentsRequestDocuments(
                path=doc.object_path,
                name=doc.path.name,
                size=doc.path.stat().st_size,
            )
            for doc in docs
        ]
        response = self.client.add_documents(
            knowledge_base_id,
            models.AddDocumentsRequest(
                knowledge_base_id=knowledge_base_id,
                import_type="LOCAL_UPLOAD",
                documents=add_docs,
                meta_fields=dict(meta_fields) if meta_fields else None,
                dedup=models.AddDocumentsRequestDedup(
                    doc_name_dedup=True,
                    content_dedup=False,
                ),
            ),
        )
        body = response.body
        self._check(body, "AddDocuments")

        errors = list(getattr(body.data, "errors", None) or [])
        if errors:
            raise RuntimeError(f"データ登録に失敗しました: {errors}")
        return body.to_map()

    def search(
        self,
        knowledge_base_id: str,
        query: str,
        options: SearchOptions,
        image_url: str | None = None,
    ) -> dict[str, Any]:
        tag_filter = None
        if options.tag_conditions:
            tag_filter = models.SearchKnowledgeBaseRequestTagFilter(
                relation=options.tag_relation,
                conditions=[
                    models.SearchKnowledgeBaseRequestTagFilterConditions(
                        field=condition.field,
                        op=condition.op,
                        value=condition.value,
                    )
                    for condition in options.tag_conditions
                ],
            )
        response = self.client.search_knowledge_base(
            knowledge_base_id,
            models.SearchKnowledgeBaseRequest(
                query=query,
                version=options.version,
                page_number=1,
                page_size=options.page_size,
                rerank_model_name=options.rerank_model_name,
                tag_filter=tag_filter,
                image=(
                    models.SearchKnowledgeBaseRequestImage(url=image_url)
                    if image_url
                    else None
                ),
                retrieval_config=models.SearchKnowledgeBaseRequestRetrievalConfig(
                    candidate_count=options.candidate_count,
                    min_score=options.min_score,
                    semantic_weight=options.semantic_weight,
                    enable_query_expansion=options.enable_query_expansion,
                ),
            ),
        )
        body = response.body
        self._check(body, "SearchKnowledgeBase")
        return body.to_map()

ステップ 5：バッチアップロードスクリプトの作成

以下の内容を upload.py として保存します。AddDocuments の MetaFields はバッチ全体に適用されるため、スクリプトはまずドキュメントをタグでグループ化し、次に 100 件ずつのバッチで送信します。

"""タグ別にローカルドキュメントをバッチアップロードします。アップロード後、コンソールで解析が完了するのを待ってからバージョンを公開します。"""

from __future__ import annotations

import argparse
import json
from collections import defaultdict
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from kb_client import KnowledgeBaseClient

@dataclass(frozen=True)
class ManifestEntry:
    path: Path
    metadata: dict[str, Any]

def load_manifest(path: Path) -> list[ManifestEntry]:
    entries: list[ManifestEntry] = []
    for line_number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
        if not raw_line.strip():
            continue
        value = json.loads(raw_line)
        file_path = (path.parent / str(value["path"])).resolve()
        metadata = value.get("metadata") or {}
        if not isinstance(metadata, dict):
            raise ValueError(f"metadata on line {line_number} of the manifest must be an object")
        entries.append(ManifestEntry(file_path, metadata))
    return entries

def discover(paths: list[str], default_metadata: dict[str, Any]) -> list[ManifestEntry]:
    entries: list[ManifestEntry] = []
    for value in paths:
        path = Path(value).expanduser()
        files = sorted(item for item in path.rglob("*") if item.is_file()) if path.is_dir() else [path]
        entries.extend(ManifestEntry(item.resolve(), default_metadata) for item in files)
    return entries

def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("paths", nargs="*", help="ファイルまたはディレクトリ。複数指定できます")
    parser.add_argument("--config", default="config.json")
    parser.add_argument("--manifest", help="JSONL ファイル。各行にはパスとメタデータが含まれます")
    args = parser.parse_args()

    config = json.loads(Path(args.config).read_text(encoding="utf-8"))
    aliyun = config["aliyun"]
    upload_config = config.get("upload") or {}
    default_metadata = upload_config.get("default_meta_fields") or {}
    entries = (
        load_manifest(Path(args.manifest).expanduser().resolve())
        if args.manifest
        else discover(args.paths, default_metadata)
    )
    if not entries:
        raise SystemExit("No files found to upload")

    grouped: dict[str, list[ManifestEntry]] = defaultdict(list)
    for entry in entries:
        key = json.dumps(entry.metadata, ensure_ascii=False, sort_keys=True)
        grouped[key].append(entry)

    client = KnowledgeBaseClient(
        aliyun["access_key_id"], aliyun["access_key_secret"], aliyun["region_id"]
    )
    submitted = 0
    for metadata_key, group in grouped.items():
        metadata = json.loads(metadata_key)
        for start in range(0, len(group), 100):
            batch = group[start : start + 100]
            client.upload(
                aliyun["knowledge_base_id"],
                [entry.path for entry in batch],
                meta_fields=metadata,
            )
            submitted += len(batch)
            print(f"Submitted {submitted}/{len(entries)} documents; metadata={metadata}")

if __name__ == "__main__":
    main()

アップロードを実行します：

python3 upload.py --manifest documents.jsonl

出力には、タグでグループ化されたバッチが表示されます。例えば、5つのドキュメントが4種類のタグの組み合わせを持つ場合、4つのバッチが生成されます：

Submitted 2/5 documents; metadata={'docType': 'policy', 'effectiveDate': '2026-01-01', 'productLine': 'all'}
Submitted 3/5 documents; metadata={'docType': 'policy', 'effectiveDate': '2026-03-01', 'productLine': 'phone'}
Submitted 4/5 documents; metadata={'docType': 'manual', 'effectiveDate': '2026-03-01', 'productLine': 'phone'}
Submitted 5/5 documents; metadata={'docType': 'faq', 'effectiveDate': '2026-02-01', 'productLine': 'all'}

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

解析が完了しても、ドキュメントは未発行のままです。バージョンを発行して初めて検索可能になります。この操作は現在、コンソールでのみ実行できます。

1. ナレッジベースの詳細ページを開き、[バージョン管理] タブをクリックし、保留中の変更があることを確認してから [バージョンを発行] をクリックします。

2. [変更の確認] ステップで、変更ログ (新しく追加されたデータが 1 件ずつ表示されます) を確認し、[Next] をクリックします。

3. [説明の入力] ステップで、リリースノート (最大 200 文字) を入力し、[Publish] をクリックします。

4. [バージョン履歴] で、新しいバージョンのステータスが [発行済み] であることを確認し、バージョン番号を記録します。最初の発行バージョンは v1 で、その後は v2、v3 と続きます。

   config.json の knowledge_base_version に LATEST_PUBLISHED を指定すると、最新の発行済みバージョンを自動的に検索します。明示的なバージョン番号を指定することもできます。安定した Q&A サービスを運用するため、DRAFT は使用しないでください。

ステップ 7：タグフィルタリングによる検索

config.json の retrieval.tag_filter.conditions にフィルター条件を設定して、検索範囲を、指定したタグに限定します。例えば、ポリシーのドキュメントのみを検索する場合：

{
  "tag_filter": {
    "relation": "and",
    "conditions": [
      {"field": "docType", "op": "=", "value": "policy"}
    ]
  }
}

relation は and (すべての条件を満たす) と or (いずれかの条件を満たす) に対応しています。op は、以下の演算子に対応しています：

実際に機能する演算子は、=、in、not in のみです。他の演算子は 400 Unsupported tag filter operator エラーの "Supported operators" リストに表示されますが、テストによると、それらを渡すと条件はサイレントに無視され、フィルターされていないすべてのデータが返されます。さらに、contains と not contains は in/not in のエイリアスにすぎず、文字列の包含ではありません。フラグメントを渡すと、0 件の結果が返されます。

これらの制限の回避策：

• 数値または日付の範囲でフィルターするには、代わりに列挙値と in を使用します。

• 空のタグをチェックするには、代わりに = "" を使用します。

• フィルター条件を設定したら、必ず合計結果件数をフィルターされていない結果と比較して、フィルターが実際に適用されていることを確認してください。

このトピックのコーパスを例に、さまざまなフィルター条件下での検索範囲：

ステップ 8：Q&A サービスの作成

以下の内容を app.py として保存します。

"""ナレッジベース検索と OpenAI 互換 LLM の Q&A サービス"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

import requests
from flask import Flask, jsonify, render_template, request

from kb_client import KnowledgeBaseClient, SearchOptions, TagCondition

CONFIG = json.loads(Path("config.json").read_text(encoding="utf-8"))
ALIYUN = CONFIG["aliyun"]
LLM = CONFIG.get("llm", {})
SCENARIO = CONFIG.get("scenario", {})
RETRIEVAL = CONFIG.get("retrieval", {})
KB = KnowledgeBaseClient(
    ALIYUN["access_key_id"], ALIYUN["access_key_secret"], ALIYUN["region_id"]
)
app = Flask(__name__)

def search_options() -> SearchOptions:
    raw_conditions = (RETRIEVAL.get("tag_filter") or {}).get("conditions") or []
    return SearchOptions(
        version=ALIYUN.get("knowledge_base_version", "LATEST_PUBLISHED"),
        page_size=int(RETRIEVAL.get("page_size", 6)),
        candidate_count=int(RETRIEVAL.get("candidate_count", 48)),
        min_score=float(RETRIEVAL.get("min_score", 0.0)),
        semantic_weight=float(RETRIEVAL.get("semantic_weight", 0.5)),
        enable_query_expansion=bool(RETRIEVAL.get("enable_query_expansion", True)),
        rerank_model_name=str(RETRIEVAL.get("rerank_model_name") or "") or None,
        tag_relation=str((RETRIEVAL.get("tag_filter") or {}).get("relation", "and")),
        tag_conditions=tuple(
            TagCondition(str(item["field"]), str(item["op"]), item.get("value"))
            for item in raw_conditions
        ),
    )

def find_results(payload: Any) -> list[dict[str, Any]]:
    """レスポンスから取得したチャンクのリストを抽出します。"""
    if isinstance(payload, list):
        return [item for item in payload if isinstance(item, dict)]
    if not isinstance(payload, dict):
        return []
    for key in ("results", "Results"):
        if isinstance(payload.get(key), list):
            return payload[key]
    for key in ("data", "Data"):
        found = find_results(payload.get(key))
        if found:
            return found
    return []

def field(item: dict[str, Any], *names: str) -> Any:
    for name in names:
        if item.get(name) not in (None, ""):
            return item[name]
    return ""

def llm_answer(question: str, results: list[dict[str, Any]]) -> str:
    if not LLM.get("enabled", True):
        return "LLM は無効です。以下の検索結果を参照してください。"
    if not results:
        return "利用可能なドキュメントには、この内容が明示的に記載されていません。"
    context = "\n\n".join(
        f"[ソース {index}] {field(item, 'documentName', 'DocumentName')}\n"
        f"{field(item, 'content', 'Content')}"
        for index, item in enumerate(results, 1)
    )
    url = str(LLM["base_url"]).rstrip("/") + "/chat/completions"
    response = requests.post(
        url,
        headers={"Authorization": f"Bearer {LLM['api_key']}"},
        json={
            "model": LLM["model"],
            "temperature": 0.1,
            "messages": [
                {"role": "system", "content": SCENARIO.get("system_prompt", "ドキュメントのみに基づいて回答してください。")},
                {"role": "user", "content": f"質問: {question}\n\n検索されたドキュメント:\n{context}"},
            ],
        },
        timeout=90,
    )
    response.raise_for_status()
    return response.json()["choices"][0]["message"]["content"].strip()

@app.get("/")
def index():
    return render_template(
        "index.html",
        title=SCENARIO.get("title", "ナレッジベース Q&A"),
        sample_questions=SCENARIO.get("sample_questions", []),
        image_enabled=bool(SCENARIO.get("image_enabled", False)),
    )

@app.post("/api/ask")
def ask():
    payload = request.get_json(silent=True) or {}
    question = str(payload.get("question", "")).strip()
    image_url = str(payload.get("image_url", "")).strip() or None
    if not question:
        return jsonify({"error": "質問は空にできません"}), 400
    try:
        raw = KB.search(
            ALIYUN["knowledge_base_id"],
            question,
            options=search_options(),
            image_url=image_url,
        )
        results = find_results(raw)
        return jsonify({"answer": llm_answer(question, results), "sources": results})
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500

if __name__ == "__main__":
    app.run(host="127.0.0.1", port=7860, debug=False)

以下の内容を templates/index.html として保存します。

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width,initial-scale=1">
  <title>{{ title }}</title>
  <style>
    body{margin:0;background:#f6f7fb;color:#1f2937;font:15px system-ui,sans-serif}
    main{max-width:860px;margin:0 auto;padding:42px 18px}
    .card{background:#fff;border:1px solid #e5e7eb;border-radius:18px;padding:24px}
    textarea{box-sizing:border-box;width:100%;min-height:100px;border:1px solid #d1d5db;border-radius:12px;padding:14px;font:inherit}
    button{margin-top:12px;border:0;border-radius:10px;padding:11px 18px;background:#4f46e5;color:#fff;cursor:pointer}
    .chip{background:#eef2ff;color:#3730a3;margin:4px;padding:7px 10px}
    pre{white-space:pre-wrap;line-height:1.65}.muted{color:#6b7280}.source{border-top:1px solid #eee;padding:12px 0}
  </style>
</head>
<body><main><h1>{{ title }}</h1><p class="muted">回答は公開されたナレッジベースのコンテンツから生成され、検索ソースが表示されます。</p>
  <div>{% for q in sample_questions %}<button class="chip" onclick='setQ({{ q|tojson }})'>{{ q }}</button>{% endfor %}</div>
  <section class="card">
    <textarea id="q" placeholder="質問を入力してください"></textarea>
    <button id="ask" onclick="ask()">送信</button>
    <pre id="answer"></pre>
    <div id="sources"></div>
  </section>
</main><script>
const q=document.querySelector('#q'), answer=document.querySelector('#answer'), sources=document.querySelector('#sources');
function setQ(value){q.value=value;q.focus()}
async function ask(){
  const text=q.value.trim(); if(!text) return;
  answer.textContent='検索・生成中です...'; sources.innerHTML='';
  const res=await fetch('/api/ask',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({question:text})});
  const data=await res.json();
  answer.textContent=data.answer||('エラー: '+data.error);
  for(const [i,item] of (data.sources||[]).entries()){
    const div=document.createElement('div'); div.className='source';
    div.textContent=`ソース ${i+1}: ${item.documentName||''}\n${item.content||''}`;
    sources.appendChild(div);
  }
}
</script></body></html>

ステップ 9：起動と検証

• サービスを起動します。

python3 app.py

• ブラウザで http://127.0.0.1:7860 を開き、サンプルの質問をクリックするか手動で入力して、[Send] をクリックします。

• API を直接呼び出して検証することもできます。

curl -sS http://127.0.0.1:7860/api/ask \
  -H 'Content-Type: application/json' \
  -d '{"question":"How long after payment does it usually take to ship the product?"}'

以下の 4 つの受け入れ基準を検証します。

• ページが正常に開き、質問を送信すると回答と検索ソースの両方が返されること。

• 回答内の各 [Source N] に、下部のソースエリアに対応するドキュメントチャンクがあること。

• 質問の内容がドキュメントに含まれていない場合、回答は事実を捏造するのではなく、「The available documents do not explicitly cover this」となること。

• コンソールでドキュメントを追加してバージョンを再度公開した後、ページで新しいコンテンツを検索できること。

チューニングの推奨事項

• ドキュメントの形状に合わせたチャンク長の調整：カスタマーサービスドキュメントは、短い FAQ エントリとポリシー条項が中心です。ナレッジベースの詳細ページの [Processing Policies] で [Create Policy] をクリックし、チャンキング方法として スマート分割 または 長さによる分割 を選択します。最大セグメント長は文字数 (デフォルト 512 文字) で計測される点に注意してください。短文中心のシナリオでは、380～580 文字 (おおよそ 256～384 トークン) に設定します。ポリシーを作成したら、インポート時に AddDocumentsRequest.strategy_id で指定します。このチュートリアルの upload.py スクリプトではこのフィールドを設定していません。カスタムの処理ポリシーを使用する場合は、AddDocuments リクエストにこのフィールドを追加してください。

• リランキングと min_score の同時調整：rerank_model_name を有効にすると、リランキングスコアとベクトル類似度スコアのスケールが一致しなくなります。元の min_score を維持すると、除外される結果が少なすぎる場合があります (テストでは、リランキングを有効にすると、3 つのカスタマーサービスの質問に対し、それぞれヒットが 1 件のみとなりました)。複数のドキュメントを組み合わせた回答が必要な質問の場合は、リランキングを有効にする際に min_score を適切に下げてください。

• 部分的に関連するドキュメントによる誤った結論への注意：質問がドキュメントと表面的には関連していても、意味的にはカバーされていない場合 (たとえば、ドキュメントが配送時間のみを説明しているのに、ユーザーが代金引換に対応しているかを質問する場合)、LLM がそこから誤った結論を導くことがあります。system_prompt で、ドキュメントが質問を直接カバーしていない場合は「分からない」と回答するよう、また部分的に関連するドキュメントからの推論を禁止するよう明示し、ローンチ前に実際の顧客質問でスポットチェックしてください。

• マニュアルのタイトルだけでなく、実際の顧客質問でのテスト：semantic_weight (例：0.7) を高めると、口語表現により良く対応できます。

• 有効な公式ポリシーを優先してインポートし、競合する古いバージョンを同時に保持しないでください。ポリシー更新後は、バージョンを再度公開し、effectiveDate タグでバージョンを区別します。

よくある質問

起動前チェックリスト

• 認証情報ストレージ： アクセスキーペアと LLM API キーは、ローカルの config.json にのみ保存してください。コードリポジトリにコミットしたり、チャットグループで共有したりしないでください。デモの完了後は、一時的な認証情報を速やかにローテーションまたは削除してください。

• 最小権限： ローテーション可能な最小権限の専用 RAM ユーザーを使用してください。Alibaba Cloud アカウントの AccessKey を長期間使用しないでください。

• 認可済みコンテンツ： 取り扱う権限のあるドキュメントのみをインポートしてください。すべての回答は、ソースチャンクまでトレースできる必要があります。

• 本番デプロイ： 外部へデプロイする場合は、Flask 開発サーバーを使い続けないでください。本番用の WSGI サーバーを使用し、認証、HTTPS、アクセスログのマスキング、レート制限、監査を追加してください。

• ヒューマンフォールバック： 重要なポリシーに関する Q&A では、ヒューマンフォールバックのエントリを用意してください。