Vector Retrieval Service for Milvus:Alibaba Cloud Milvus を使用した顧客サービス音声分析および FAQ マッチングパイプラインの構築

ソリューション概要

コンタクトセンターチームは日々、通話録音や音声メッセージを蓄積していますが、ほとんどのチームはこのデータをコンプライアンスのためのアーカイブとしてのみ保持しています。録音には、顧客のリクエスト、エージェントのサービス品質の証拠、否定的なフィードバックの初期兆候が含まれています。この音声資産を有効活用するには、以下の機能が必要です：

• 音声テキスト変換 — 通話や音声メッセージをまとめてテキストに文字起こしし、検索や分析を可能にします。他のすべてがこのステップに依存します。

• 検索可能なナレッジベース — 質の高い過去の返信、製品マニュアル、FAQ を、ボットやエージェントがリアルタイムでクエリできるセマンティック検索可能なナレッジベースに変換します。

• インテリジェントな FAQ マッチング — キーワードの一致に頼るのではなく、顧客の口語的な質問を正しい標準 FAQ にマッチングさせます。

• 感情およびインテント認識 — 通話中の感情の変化を検出し、リアルタイムのアラートや通話後の品質検査をサポートします。

• 品質検査とコンプライアンスマスキング — 文字起こしされたテキストには、携帯電話番号、ID カード番号、銀行カード番号などの個人識別情報 (PII) が含まれることがよくあります。このデータは、保存、分析、共有の前にマスキングします。

従来のアプローチでこの一連の機能を組み立てるには、通常、ASR サービス、ベクトルデータベース、外部の NLP プラットフォーム、再ランキングサービスという 4 つ以上のシステムを連結する必要があります。グルーコードは増え続け、いずれか 1 つのホップの不安定さがパイプライン全体を低速化させます。生の文字起こしテキストを外部サービスに送信すると、PII が漏洩するリスクもあります。

Alibaba Cloud Milvus の AI 機能は、これらの機能を単一のベクトルデータベースに統合します。Milvus は、書き込み時と検索時に内部でモデル推論をトリガーし、データがインスタンスから外部に出ることはありません。次の表では、このチュートリアルで使用する 6 つの AI 機能について説明します。

AI 機能とマルチシステムスタックとの比較

次の表は、このパイプラインと、個別のサービスから組み立てられた同等のパイプラインを比較したものです。

前提条件

• Milvus インスタンス — Milvus 2.6 インスタンス。AI 機能には 2.6 カーネルが必要であり、作成後に別途モデルサービスをバインドする必要はありません。

• インターネットアクセス — インスタンスにインターネット経由でアクセスするには、インスタンス詳細ページの [Security Configuration] タブで [Public Endpoint] を有効にし、クライアントの出口 IP アドレスをパブリックアクセスホワイトリストに追加します。

• エンドポイントポート — インスタンス URI にポート 19530 を明示的に指定します (例： http://c-xxx.milvus.aliyuncs.com:19530)。RESTful API と gRPC はこのポートを共有します。ポートを省略すると、接続はポート 80 にフォールバックし、タイムアウトします。

• クライアントライブラリ — pymilvus がインストールされていること。このトピックの例は、pymilvus 3.0.0 で検証されています。

• 音声ソース — 文字起こしする録音が、サーバーが実際にダウンロードできるアドレスにアップロードされていること。本番環境では、ご自身の Object Storage Service (OSS) バケットから取得した有効期間の短い署名付き URL を使用します。プレースホルダーや有効期限切れのアドレスでは、文字起こし時に失敗します。

• メディア認証情報 — すべてのメディアに対して、有効期間が短く、最小権限の署名付き URL を使用します。URL に長期的な認証情報を埋め込まないでください。

• 録音の同意 — 録音を処理する前に、録音に関する開示と顧客の同意が完了していること。

共有コードの準備

次のコードには、接続設定、REST コールラッパー、および TEXTTRANSFORM 関数タイプの互換性フォールバックが含まれています。すべてのステップで再利用されます。このセクションは準備のみであり、6 つのステップのいずれにもカウントされません。

MILVUS_URI と MILVUS_TOKEN をお使いのインスタンスの値に置き換えます。

from __future__ import annotations

import json
import time
from typing import Any
from urllib.error import HTTPError
from urllib.request import Request, urlopen

from pymilvus import DataType, Function, FunctionType, MilvusClient

# ==================== 接続設定 ====================
MILVUS_URI = "http://c-xxx.milvus.aliyuncs.com:19530"  # ポートは 19530 を指定する必要があります
MILVUS_TOKEN = "root:xxx"
MILVUS_REST_BASE_URL = MILVUS_URI

client = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN)

# AI_AUDIO_TRANSCRIBE / AI_PII_MASK / AI_SENTIMENT / AI_CLASSIFY はすべて
# TEXTTRANSFORM 型 (関数型値 9) の関数であり、task パラメーターによって区別されます。
# 一部の pymilvus バージョンの FunctionType enum にはそのようなメンバーがないため、フォールバックを追加します。
TEXTTRANSFORM_FUNCTION_TYPE = 9

def texttransform_function_type() -> Any:
    for type_name in ("TEXTTRANSFORM", "TEXT_TRANSFORM", "TextTransform"):
        ft = getattr(FunctionType, type_name, None)
        if ft is not None:
            return ft
    existing = getattr(FunctionType, "_value2member_map_", {}).get(TEXTTRANSFORM_FUNCTION_TYPE)
    if existing is not None:
        return existing
    extension = int.__new__(FunctionType, TEXTTRANSFORM_FUNCTION_TYPE)
    extension._name_ = "TEXTTRANSFORM"
    extension._value_ = TEXTTRANSFORM_FUNCTION_TYPE
    FunctionType._value2member_map_[TEXTTRANSFORM_FUNCTION_TYPE] = extension
    FunctionType._member_map_["TEXTTRANSFORM"] = extension
    return extension

def post_json(path: str, body: dict[str, Any], timeout: int = 120,
              retries: int = 3) -> tuple[int, dict[str, Any]]:
    """同期 REST API の共有ラッパー。これらの呼び出しは LLM を経由するため、リトライによって安定性が向上します。"""
    last: tuple[int, dict[str, Any]] | None = None
    for _ in range(retries):
        request = Request(
            f"{MILVUS_REST_BASE_URL.rstrip('/')}{path}",
            data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
            headers={"Authorization": f"Bearer {MILVUS_TOKEN}",
                     "Content-Type": "application/json"},
            method="POST",
        )
        try:
            with urlopen(request, timeout=timeout) as response:
                status, data = response.status, json.loads(response.read().decode("utf-8"))
        except HTTPError as exc:
            status, data = exc.code, json.loads(exc.read().decode("utf-8"))
        last = (status, data)
        if status == 200 and data.get("code") == 0:
            return status, data
        time.sleep(1)
    return last

いずれかのステップを実行する前に、共有コードの以下の特性に留意してください：

• 関数タイプ — AI_AUDIO_TRANSCRIBE、AI_PII_MASK、AI_SENTIMENT、および AI_CLASSIFY はすべて TEXTTRANSFORM タイプの関数であり、task パラメータによって区別されるのに対し、AI_EMBEDDING と AI_RERANK は独自の FunctionType 定数を持ちます。

• リトライ範囲 — post_json は、レスポンスが HTTP エラーであるか、ペイロードがゼロ以外の code を返す場合に、呼び出しを最大 3 回リトライします。読み取りタイムアウトやネットワークエラーはキャッチされず、呼び出し元に伝播します。

• ステップの独立性 — 各ステップは独自のコレクションを作成し、独立して実行されます。ステップを連結するには、あるステップの出力フィールドを次のステップの入力フィールドに渡します。ステップ 1 の transcript フィールドはステップ 2 のマスキング入力となり、マスキングされたテキストはステップ 6 の content 入力となります。

ステップ 1：通話録音の文字起こし

通話録音をまとめてテキストに文字起こしします。AI_AUDIO_TRANSCRIBE をコレクションにアタッチすると、音声アドレスを書き込むだけで、アプリケーションからの ASR 呼び出しなしに、文字起こしテキストが自動的に生成されます。

# ==================== ステップ 1：AI_AUDIO_TRANSCRIBE による文字起こし ====================
collection_name = "cs_transcribe"
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("audio_url", DataType.VARCHAR, max_length=4096)   # 音声入力フィールド
schema.add_field("transcript", DataType.VARCHAR, max_length=4096)  # 文字起こし出力フィールド
# コレクションには少なくとも 1 つのベクトルフィールドが含まれている必要があります。このステップではベクトル検索を実行しないため、
# 2 次元のプレースホルダーフィールドで制約を満たします。
# nullable=True は、挿入時にフィールドを省略できることを意味します。
schema.add_field("dummy_vector", DataType.FLOAT_VECTOR, dim=2, nullable=True)
schema.add_function(
    Function(
        name="transcribe_audio",
        function_type=texttransform_function_type(),
        input_field_names=["audio_url"],
        output_field_names=["transcript"],
        params={
            "provider": "aliyun_milvus",
            "model_name": "qwen3-asr-flash",
            "task": "ai_audio_transcribe",
            "language": "zh",
            "enable_itn": "true",     # 「いちさんぱち」のような話し言葉の数字を「138」に正規化します
        },
    )
)

index_params = client.prepare_index_params()
index_params.add_index(field_name="dummy_vector", index_type="AUTOINDEX", metric_type="COSINE")
client.create_collection(collection_name=collection_name, schema=schema,
                        index_params=index_params)

# audio_url は、サーバーがダウンロードできる実際のアドレスである必要があります。本番環境では、
# 独自の OSS バケットからの有効期間の短い署名付き URL を使用します。
audio_urls = [
    "https://<your-bucket>.oss-cn-hangzhou.aliyuncs.com/calls/call_0001.mp3",
    "https://<your-bucket>.oss-cn-hangzhou.aliyuncs.com/calls/call_0002.wav",
]
client.insert(collection_name, [{"audio_url": u} for u in audio_urls])
client.flush(collection_name)

for row in client.query(collection_name, filter="",
                       output_fields=["audio_url", "transcript"], limit=10):
    print(f"{row['audio_url'].split('/')[-1]} -> {row['transcript']}")

この例の 2 つのパラメーターが文字起こしの動作を制御します：

• language — この例では zh に設定されているため、中国語の録音を文字起こしします。

• enable_itn — 逆テキスト正規化を有効にします。これにより、後の検索や構造化処理が簡素化されます。

  プレースホルダーのベクトルフィールドとその nullable=True 宣言は両方とも必須です。プレースホルダーフィールドがないと、コレクションの作成は失敗します。nullable=True がないと、フィールドを省略した insert は失敗します。正確なエラーメッセージについては、「トラブルシューティング」をご参照ください。

最終的なクエリは、録音ごとに 1 行、call_0001.mp3 -> <transcript> の形式で出力します。文字起こしテキストが空の場合は、ステップ 2 に進む前に音声アドレスを確認してください。

ステップ 2：保存前の PII のマスキング

文字起こしされたテキストには、携帯電話番号、ID カード番号、銀行カード番号などの個人識別情報 (PII) が含まれることがよくあります。テキストをナレッジベースや分析ストアに入れる前にマスキングすることで、コンプライアンスをパイプラインの最後ではなく最初に移動させます。次のいずれかのパターンを選択してください：

• (このパイプラインで推奨) 同期 REST API — データが流入する前の一括処理に使用します。マスキングされたテキストのみがデータベースに到達し、マスキングされた文字起こしテキストは直接ステップ 6 に渡されます。

• 書き込み時コレクション — 監査のために、マスキングされた結果を元のテキストと一緒に保持する必要がある場合に使用します。

同期 REST API を呼び出して、テキストがどこにも書き込まれる前にマスキングします：

# ==================== ステップ 2：AI_PII_MASK によるマスキング ====================
# 2.1 同期 REST API：データベースに入る前にテキストをマスキング
status, data = post_json(
    "/v2/vectordb/ai/pii_mask",
    {
        "model_name": "qwen3.7-max",
        "texts": ["Hello, my phone number is 13800138000 and my ID card number is 110101199001010000. Please check my ticket."],
        "params": {
            "pii_types": ["PERSON", "PHONE", "ID_CARD"],
            "mask_char": "*",
            "preserve_length": True,   # マスキング後も元の長さを維持
            "temperature": 0,
        },
    },
)
assert status == 200 and data.get("code") == 0, data
for item in data["data"]["output"]["outputs"]:
    print(item)

preserve_length を true に設定すると、マスクされた値は元の文字長を維持し、元の値を公開することなくフォーマットの特性を保持します。上記の REST コールでは、以下の出力が確認されます。

Hello, my phone number is *********** and my ID card number is ******************. Please check my ticket.

11 桁の携帯電話番号は 11 個のアスタリスクとして、18 桁の ID カード番号は 18 個のアスタリスクとしてマスキングされます。

代替案として、AI_PII_MASK をコレクションにアタッチすると、書き込み時にマスキングされたフィールドが自動的に生成されます：

# 2.2 書き込み時コレクション：コンテンツが書き込まれると、マスキングされたフィールドが自動的に生成される
collection_name = "cs_pii_mask"
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("content", DataType.VARCHAR, max_length=4096)
schema.add_field("masked", DataType.VARCHAR, max_length=4096)
schema.add_field("dummy_vector", DataType.FLOAT_VECTOR, dim=2, nullable=True)
schema.add_function(
    Function(
        name="mask_pii",
        function_type=texttransform_function_type(),
        input_field_names=["content"],
        output_field_names=["masked"],
        params={
            "provider": "aliyun_milvus",
            "model_name": "qwen3.7-max",
            "task": "ai_pii_mask",
            "pii_types": "PERSON,PHONE,ID_CARD",
            "mask_char": "*",
            "preserve_length": "true",
            "temperature": "0",
        },
    )
)
index_params = client.prepare_index_params()
index_params.add_index(field_name="dummy_vector", index_type="AUTOINDEX", metric_type="COSINE")
client.create_collection(collection_name=collection_name, schema=schema,
                        index_params=index_params)

client.insert(collection_name,
              [{"content": "Hello, my phone number is 13800138000. Please check my ticket."}])
client.flush(collection_name)
for row in client.query(collection_name, filter="", output_fields=["masked"], limit=1):
    print(row["masked"])

この代替案では、携帯電話番号のみを含む、より短いサンプルテキストが書き込まれるため、その masked 値は、前に示した REST 出力とは異なります。出力された値で文の構造が維持され、携帯電話番号の数字がアスタリスクに置き換えられていることを確認してください。

ステップ 3：FAQ ナレッジベースの構築

FAQ の質問とそれに対する標準回答をコレクションに書き込みます。AI_EMBEDDING が書き込み時にこれらをベクトル化するため、アプリケーションは最初に埋め込みモデルを呼び出す必要はありません。

# ==================== ステップ 3：AI_EMBEDDING による FAQ ナレッジベースの構築 ====================
collection_name = "cs_faq_kb"
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("content", DataType.VARCHAR, max_length=4096)   # FAQ 質問テキスト
schema.add_field("answer", DataType.VARCHAR, max_length=4096)    # 標準回答
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=1024)
schema.add_function(
    Function(
        name="embed_content",
        function_type=FunctionType.TEXTEMBEDDING,
        input_field_names=["content"],
        output_field_names=["embedding"],
        params={
            "provider": "aliyun_milvus",
            "model_name": "text-embedding-v4",
            "dim": 1024,
            "max_client_batch_size": 10,
            "max_concurrency": 1,
        },
    )
)
index_params = client.prepare_index_params()
index_params.add_index(
    field_name="embedding",
    index_type="HNSW",
    metric_type="COSINE",
    params={"M": 16, "efConstruction": 200},
)
client.create_collection(collection_name=collection_name, schema=schema,
                        index_params=index_params)

faqs = [
    {"content": "How do I enable public network access for a Serverless Milvus instance?",
     "answer": "Enable public network access on the instance details page in the console and configure the whitelist."},
    {"content": "What do I do if I forget the console logon password?", "answer": "Reset it through the password recovery process in Account Center."},
    {"content": "Why is my bill higher than expected?", "answer": "Check the usage details, with a focus on compute and storage usage."},
    {"content": "How do I create a Collection and insert vectors?",
     "answer": "Define the schema with create_collection, then call insert."},
    {"content": "Does scaling up an instance affect online workloads?", "answer": "Scaling up is an online operation and usually does not interrupt service."},
]
client.insert(collection_name, faqs)
client.flush(collection_name)
client.load_collection(collection_name)

この例では、多言語とカスタムディメンションをサポートする text-embedding-v4 を使用します。 AI Center は、qwen3.7-text-embedding などのモデルも提供しています。 コンソールの AI Center の [モデルサービス] タブで利用可能なモデルとそのディメンションを確認し、言語カバレッジと品質要件に基づいてモデルを選択してください。 どのモデルを選択しても、ベクトルフィールドの dim は、関数パラメーターの dim と一致させる必要があります。

5 つすべての FAQ エントリが書き込まれ、load_collection が返るまで、ステップ 4 に進まないでください。 ロードされていないコレクションに対する検索では、ベクトル化が成功したかどうかを確認できません。

ステップ 4：セマンティック検索による FAQ 候補の取得

顧客の口語的な質問を FAQ ナレッジベースに送信し、上位 N 件の候補を取得します。ベクトル検索だけでは、広範ではあるものの正確な取得はできないため、この出力を最終的な回答ではなく候補セットとして扱ってください。

# ==================== ステップ 4：セマンティック検索による FAQ の取得 ====================
query = "How can I get my instance connected from the internet?"
result = client.search(
    collection_name="cs_faq_kb",
    data=[query],
    anns_field="embedding",
    limit=3,
    output_fields=["content", "answer"],
)
for rank, hit in enumerate(result[0], 1):
    print(f"{rank}. [similarity {hit['distance']:.4f}] {hit['entity']['content']}")

出力には、3 つの FAQ 候補が COSINE 類似度スコアとともに、高いものから低いものへとランク付けされてリストされます。このランキングを控えておいてください。ステップ 5 では、同じクエリに対する再ランキング後のランキングと比較します。

ステップ 5：候補の再ランキング

再ランキングモデルは、クエリに対する実際の関連度に基づいて候補を並べ替えます。検索と再ランキングを組み合わせることによってのみ、標準 FAQ を確実にマッチングできます。次のいずれかのパターンを選択してください：

• (このパイプラインで推奨) search にアタッチされたランカー — 検索と詳細なランキングを 1 回の呼び出しに統合し、パイプラインを短縮します。

• 同期 REST API — 候補リストが既に存在し、再ランキングのみが必要な場合に使用します。

# ==================== ステップ 5：AI_RERANK による再ランキング ====================
# 5.1 同期 REST API：指定された候補のバッチを独立して再ランキング
status, data = post_json(
    "/v2/vectordb/ai/rerank",
    {
        "model_name": "qwen3-rerank",
        "query": query,
        "documents": [
            "How do I enable public network access for a Serverless Milvus instance?",
            "Does scaling up an instance affect online workloads?",
            "How do I create a Collection and insert vectors?",
        ],
        "params": {"max_concurrency": 2, "timeout_sec": 10},
    },
)
assert status == 200 and data.get("code") == 0, data
ranked = sorted(data["data"]["output"]["results"],
                key=lambda x: x["relevance_score"], reverse=True)
for rank, item in enumerate(ranked, 1):
    print(f"{rank}. [relevance {item['relevance_score']:.4f}] candidate index={item['index']}")

# 5.2 ランカーを search にアタッチし、検索と再ランキングを 1 回の呼び出しで完了させる
reranker = Function(
    name="rerank_faq",
    function_type=FunctionType.RERANK,
    input_field_names=["content"],
    params={
        "reranker": "model",
        "provider": "aliyun_milvus",
        "model_name": "qwen3-rerank",
        "queries": [query],
        "max_concurrency": 2,
        "timeout_sec": 10,
    },
)
result = client.search(
    collection_name="cs_faq_kb",
    data=[query],
    anns_field="embedding",
    limit=3,
    output_fields=["content", "answer"],
    ranker=reranker,
)
for rank, hit in enumerate(result[0], 1):
    e = hit["entity"]
    print(f"{rank}. [rerank score {hit['distance']:.4f}] {e['content']} | answer: {e['answer']}")

REST API は top_n をサポートしておらず、候補ごとに 1 つのスコアを返します。切り捨てが必要な場合は、スコアでソートし、アプリケーション側で上位 N 件を取得してください。

クエリ「How can I get my instance connected from the internet?」に対して、以下のデータが確認できます。

2 つのスコア列は同じスケールではありません。ステップ 4 の列はベクトル間の COSINE 類似度を示し、ステップ 5 の列は再ランキングモデルの関連度スコアを示します。絶対値ではなく、ランクを比較してください。

ベクトル検索では、「What do I do if I forget the console logon password?」が 0.5613 で 2 位にランク付けされましたが、そのインテントはインターネットからの接続とは何の関係もありません。これは、純粋なベクトル検索の典型的な弱点です。表面的なテキストの類似性は高いままでも、実際のインテントは乖離しています。再ランキング後、その候補は 3 位に下がり、意味的により近いインスタンスのスケーリングに関する質問が 2 位に上がりました。顧客サービスボットにとって、Top-1 の精度は回答の品質を直接決定するため、再ランキングの段階は省略できません。

テキストのみのリランキングでは、is_multimodal パラメーターは必要ありません。そのパラメーターは、qwen3-vl-rerank を使用して画像および動画の候補のスコア付けを行うマルチモーダルリランキングにのみ適用されます。テキストのみの候補に対してこのパラメーターを渡しても、スコアは変更されません。

ステップ 6：感情およびチケットカテゴリのラベリング

会話テキストに、品質検査と否定的なフィードバックのアラートのために感情を判断する関数と、リクエストの自動配信のためにチケットのカテゴリを割り当てる関数の 2 つをアタッチします。これらの関数は両方とも、書き込み時に自動的に実行されます。完全なパイプラインでは、content フィールドはステップ 2 で生成されたマスクされたテキストを受け取りますが、次の例では単独で実行されるように生のテキストを書き込みます。

# ==================== ステップ 6：AI_SENTIMENT による感情分析 + AI_CLASSIFY によるチケット分類 ====================
collection_name = "cs_qc"
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("content", DataType.VARCHAR, max_length=4096)     # マスキングされた会話テキスト
schema.add_field("sentiment", DataType.VARCHAR, max_length=64)     # 感情出力
schema.add_field("category", DataType.VARCHAR, max_length=64)      # チケットカテゴリ出力
schema.add_field("dummy_vector", DataType.FLOAT_VECTOR, dim=2, nullable=True)
schema.add_function(
    Function(
        name="analyze_sentiment",
        function_type=texttransform_function_type(),
        input_field_names=["content"], output_field_names=["sentiment"],
        params={"provider": "aliyun_milvus", "model_name": "qwen3.7-max",
                "task": "ai_sentiment", "categories": "positive,negative,neutral",
                "temperature": "0"},
    )
)
schema.add_function(
    Function(
        name="classify_ticket",
        function_type=texttransform_function_type(),
        input_field_names=["content"], output_field_names=["category"],
        params={"provider": "aliyun_milvus", "model_name": "qwen3.7-max",
                "task": "ai_classify", "labels": "account,inquiry,fault,billing",
                "prompt": "顧客の質問のトピックに基づいて分類します。", "temperature": "0"},
    )
)
index_params = client.prepare_index_params()
index_params.add_index(field_name="dummy_vector", index_type="AUTOINDEX", metric_type="COSINE")
client.create_collection(collection_name=collection_name, schema=schema,
                        index_params=index_params)

client.insert(collection_name, [
    {"content": "I have called three times about this problem and it is still not fixed. This is very disappointing!"},
    {"content": "How do I enable public network access for a Serverless Milvus instance?"},
])
client.flush(collection_name)
for row in client.query(collection_name, filter="",
                        output_fields=["content", "sentiment", "category"], limit=10):
    print(f"sentiment={row['sentiment']:<10} category={row['category']:<6} | {row['content'][:28]}")

2 つのサンプル会話に対して、以下の結果が確認できます：

出力フィールドが競合しない限り、1 つのコレクションで複数の関数を実行できます。この例では、2 つの関数が sentiment と category を個別に設定するため、1 回の書き込みでセンチメントラベルとチケットカテゴリの両方が生成されます。

トラブルシューティング

次の表は、このチュートリアルの例で最も頻繁に返されるエラーの一覧です。

クリーンアップ

この例では、cs_transcribe、cs_pii_mask、cs_faq_kb、cs_qc の 4 つのコレクションを作成します。これらのコレクションとそのインデックスはインスタンスに残り、cs_faq_kb は削除されるまでロードされたままになります。

チュートリアルが完了したら、コレクションを削除してください：

for name in ("cs_transcribe", "cs_pii_mask", "cs_faq_kb", "cs_qc"):
    if client.has_collection(name):
        client.drop_collection(name)

クリーンアップが完了したと見なす前に、4 つの名前のそれぞれについて client.has_collection(name) が False を返すことを確認します。

データ保持とコンプライアンス

録音データは、パイプラインが終了した後も機密情報として扱われます。保存されたアーティファクトは次のように処理してください：

• 保持期間 — 生の音声、文字起こしテキスト、感情ラベル、レビュー記録に対して、明確な保持期間を設定します。期間が終了したら、これらを削除または匿名化します。

• 派生データ — 派生データは、ソースデータの削除または同意の撤回と同時に処理します。

• モデル出力のレビュー — 感情分析と分類の結果は、影響の大きいアクションを実行する前に、人によるレビュー段階が必要です。詳細については、「ステップ 6：感情およびチケットカテゴリのラベリング」をご参照ください。

  録音を処理する前に、同意、開示、署名付き URL の要件が適用されます。詳細については、「前提条件」をご参照ください。

次のステップ

以下のシナリオは、このパイプラインを拡張するものです：

• リアルタイムエージェントアシスト — 検索と再ランキングをエージェントのワークスペースに接続し、実際の通話中に最も関連性の高いスクリプトやナレッジをプッシュします。

• バッチ品質検査 — 過去の録音と AI_BATCH を組み合わせてオフラインで全量の文字起こし、マスキング、感情ラベリングを行うことで、品質検査カバレッジをサンプリング検査から全量検査に引き上げます。AI_BATCH はこのチュートリアルの範囲外です。

• ネガティブフィードバックアラート：AI_SENTIMENT の出力が negative であるチケットに対して、リアルタイムでアラートとエスカレーションをトリガーします。