【Azure AI Foundry：第3回】知識のセキュア連携：Azure AI SearchとのRAGエージェント構築

【新連載：Azure AI Foundryとマルチエージェント】

前回（第2回）：【Azure AI Foundry：第2回】知能のステート管理：Threadによる会話履歴の自動永続化

※本連載は、Azure AI Foundry SDK (azure-ai-projects) を用い、実務で安全・強力に動作する「エンタープライズ仕様の自律チームエージェント」を構築するステップバイステップのハンズオン連載です。

1. イントロダクション（今回の到達目標）

前回は、会話の履歴をクラウド側で永続化する Thread 機能を実装し、エージェントに「文脈を維持する記憶」を与えました。

しかし、エージェントがビジネスの実務で成果を出すためには、一般的な対話能力だけでは不十分です。LLMのトレーニングデータには含まれていない「自社独自の仕様書、製品マニュアル、契約書」といったプライベートな知識をもとに、正確で嘘（ハルシネーション）のない回答を行わせる必要があります。

今回は、エンタープライズ向けの検索エンジンである「Azure AI Search」に格納されたドキュメント情報を、SDKを介してエージェントへ安全かつ直接的に接続し、質問に対して適切な知識を意味検索（セマンティック検索）して回答の根拠（引用元）と共に提示する「ドキュメント検索エージェント」を構築することを目標とします。

前提条件（Prerequisites）

動作確認済みライブラリ: azure-ai-projects>=2.0.0
Azure上の追加リソース：
- Azure AI Search のプロビジョニングと、検証用データの取り込み（インデックスの作成）が完了していること。
- Azure AI Foundryプロジェクトから、該当する Azure AI Search に対する「接続（Connection）」が設定されており、その接続IDが取得可能であること。
- 認証ユーザーに、Azure AI Search リソースへの Search Index Reader および Search Traffic Contributor などのロールが割り当てられていること。

2. なぜ「Azure AI Search直接連携」なのか

一般的なRAG（Retrieval-Augmented Generation）開発では、以下のような「つなぎ込みのコード」を開発者がすべて自前で記述していました。

ユーザーの質問から埋め込みベクトルを作成する。
ベクトルデータベースへ検索クエリを投げ、関連テキストを抽出する。
抽出した結果をシステムプロンプトのコンテキスト部に整形して結合し、LLMへ送信する。

この実装は、データベースへの直接アクセス資格情報の管理や、長いコンテキストの構築ロジックをクライアント側で抱え込む必要があり、セキュリティ境界の担保や保守性の面で課題がありました。

Azure AI Foundry の Agent サービスでは、「AzureAISearchTool」を利用することで、これらの接続処理をすべてクラウド側（データプレーン内）で完結させることができます。

エージェントを作成する際、対象の「接続名」と「インデックス名」を指定して AzureAISearchTool リソースとして登録するだけで、エージェント自身が必要な時に自動でデータベースを検索（RAG）し、結果を裏で分析して回答を組み立てる自律的な知識アクセスモデルが完成します。

3. 【実践】Azure AI Searchと連携したRAGエージェントの構築

接続済みの Azure AI Search インデックスから情報を動的に検索し、そのデータに基づいて根拠（引用情報）付きで回答を生成する完全な Python スクリプトを実行してみましょう。

ローカル環境変数の AI_SEARCH_CONNECTION_NAME と AI_SEARCH_INDEX_NAME を正しく設定した上で実行します。

# 動作確認済みライブラリバージョン: azure-ai-projects==2.0.0
import os
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import (
    AzureAISearchTool,
    AzureAISearchToolResource,
    AISearchIndexResource,
    AzureAISearchQueryType
)

def main():
    endpoint = os.environ.get("AZURE_AI_PROJECT_ENDPOINT")
    connection_name = os.environ.get("AI_SEARCH_CONNECTION_NAME")  # ポータルで登録したAI Searchの接続名
    index_name = os.environ.get("AI_SEARCH_INDEX_NAME")            # AI Searchのインデックス名

    if not all([endpoint, connection_name, index_name]):
        print("Error: 環境変数 AZURE_AI_PROJECT_ENDPOINT, AI_SEARCH_CONNECTION_NAME, AI_SEARCH_INDEX_NAME のいずれかが設定されていません。")
        return

    credential = DefaultAzureCredential()

    with AIProjectClient(endpoint=endpoint, credential=credential) as project_client:
        
        # 1. 接続名からプロジェクト接続IDを取得
        print(f"🔍 接続 '{connection_name}' からプロジェクト接続IDを検索中...")
        connection = project_client.connections.get(connection_name)
        project_connection_id = connection.id
        print(f"✅ 接続ID取得成功 (ID: {project_connection_id})")

        # 2. AzureAISearchTool の構成定義
        # 最新のSDK（2.0.0以上）では AzureAISearchTool を使用し、接続とインデックスを指定します
        search_tool = AzureAISearchTool(
            azure_ai_search=AzureAISearchToolResource(
                indexes=[
                    AISearchIndexResource(
                        project_connection_id=project_connection_id,
                        index_name=index_name,
                        query_type=AzureAISearchQueryType.SIMPLE  # または SEMANTIC
                    )
                ]
            )
        )

        # 3. ナレッジツールを装備した「Document Agent」の作成
        print("\n🤖 Document Agent を作成中...")
        agent = project_client.agents.create_agent(
            model="gpt-4o",
            name="document-search-agent",
            instructions="あなたは社内ドキュメント検索の専門家です。質問に対して、提供された Azure AI Search インデックスから検索し、必ずそのドキュメントの内容を根拠として回答してください。",
            tools=search_tool.definitions,
            tool_resources=search_tool.resources
        )
        print(f"✅ エージェント作成完了 (ID: {agent.id})")

        # 4. 会話スレッドの開始と質問の送信
        thread = project_client.agents.create_thread()
        
        user_query = "プロジェクトXの現在のリリーススケジュールと、主要なマイルストーンを教えてください。"
        print(f"\n➡️ ユーザー: {user_query}")
        
        project_client.agents.create_message(
            thread_id=thread.id,
            role="user",
            content=user_query
        )

        # 5. 実行のトリガー
        print("⚡ システム: エージェントが Azure AI Search から検索を実行中...")
        run = project_client.agents.create_and_process_run(
            thread_id=thread.id,
            assistant_id=agent.id
        )
        print(f"✅ 実行ステータス: {run.status}")

        # 6. メッセージ一覧と引用（Citations）の確認
        messages = project_client.agents.list_messages(thread_id=thread.id)
        
        print("\n=== AIからの返答 (RAG検索結果ベース) ===")
        # 最新のメッセージ（直近のAIの回答）を取得して表示します
        latest_message = messages.data[0]
        for part in latest_message.content:
            if hasattr(part, 'text'):
                text_value = part.text.value
                # 引用メタデータがある場合は表示
                if hasattr(part.text, 'annotations') and part.text.annotations:
                    print(text_value)
                    print("\n📚 【引用文献ソース】:")
                    for anno in part.text.annotations:
                        # 引用ファイルの名称や箇所を出力
                        if hasattr(anno, 'file_citation'):
                            print(f" -> ファイル参照ID: {anno.file_citation.file_id}")
                else:
                    print(text_value)
        print("========================================")

        # クリーンアップ
        project_client.agents.delete_agent(agent.id)
        print("\n🧹 リソースをクリーンアップしました。")

if __name__ == "__main__":
    main()

4. まとめと次回の展望

今回は、AzureAISearchTool の接続定義をエージェントに直接装備することで、クライアント側で検索ロジックやデータベースの認証情報を抱え込むことなく、安全に「ドキュメント検索（RAG）」を実行して回答する知能を実装しました。

これで、エージェントは過去の会話を覚えた上で、自社独自のナレッジベースから必要な情報を引き出す賢さを手に入れました。

しかし、実務の中では、抽出した膨大な売上数値やログデータを「集計してレポートを作成する」あるいは「グラフ画像としてプロットする」といった、計算機としての処理が必要な局面が数多く存在します。テキストしか出力できないLLMにとって、これらは非常に苦手な領域です。

次回、第4回。
エージェント自身にPythonのコードを自動生成させ、Azure側で完全に隔離された安全な仮想サンドボックス内でコードを動的実行させて集計グラフを出力する、「Managed Code Interpreterでのデータ分析」を彼に装着します。

言葉を超えて、知能が「プログラムの実行力」という強力な物理アームを獲得する瞬間へと進みましょう。