Azure AI SearchのAgentic Retrievalを使ってAgentic RAG実現

こんにちは、2年目エンジニアの飯棲です。
すっかり暖かくなり春本番ですね。
オフィス近くの公園に咲く桜がとても綺麗でした。

今回は、Microsoft Azureのプレビューとして新たに追加されたAgentic Retrieval について紹介します。
これにより、RAG を備えたエージェントを簡単に作成できるようになりました。

1. はじめに

Azure AI Searchにて、Agentic Retrieval機能を使うことができるようになりました。
この機能は2025年5月に preview として公開されたものです。

learn.microsoft.com

Agentic Retrieval は、ユーザーの入力から必要なクエリを自律的に作成し、検索した結果を集約して返答することができます。
複雑な質問であっても複数回に分けて自動的に必要な知識を検索することが可能です。

図の引用：エージェント取得の概要 - Azure AI Search | Microsoft Learn

2. AI SearchのAgentic Retrievalについて

2.1 何ができるか

Agentic Retrievalを使うことで、AI SearchのナレッジベースとAzure AI Foundryのモデルを使って、登録したファイルを検索して回答するエージェントを作ることができます。
エージェントが検索の必要性を判断したタイミングで検索を実行し、回答への組み込みまで行ってくれます。
このとき、検索結果の並び替えには semantic rerank もあわせて行われます。

Agentic Retrievalを使う際は、Output Mode, Reasoning Effortの2つのパラメータで出力する内容、検索時の思考の深さを設定できます。

2.2 Output Mode

検索時には、outputMode というパラメータによって、retrieve 実行時に「抽出データだけ返すか」「回答文まで生成して返すか」を切り替えます。
outputModeには ANSWER_SYNTHESIS と EXTRACTIVE_DATA の2種類があります。

• ANSWER_SYNTHESIS: 取得した根拠をもとに、LLMが自然文の回答を生成して返すモード。
• EXTRACTIVE_DATA: 根拠データ（チャンク）をそのまま返すモード。

検索時に回答文まで作成したいときにはANSWER_SYNTHESISを、自前で取得結果の後処理を使いたい場合はEXTRACTIVE_DATAを使います。

2.3 Reasoning Effort

Agentic Retrievalでは、検索時の思考量を調整するReasoning Effortパラメータを設定可能です。

learn.microsoft.com

Reasoning Effortには、minimal, low, mediumの3種類があります。

minimal は LLM ベースのクエリ計画を使わず、指定した knowledge source に対して直接テキスト/ベクトル検索を実行します。
LLMなしの従来通りの検索を行うモードであり、ANSWER_SYNTHESIS は使えません。

一方で、low, mediumはLLMを介した検索を行い、ユーザーの質問に対して回答に必要な情報を得るためのサブクエリを生成し、検索を実行します。
mediumの方が目的の情報源が得られるまで深い探索を行い、回答を得るための検索回数がより増える傾向にあります。

3. 試してみる

3.1 今回やること

今回は、Blob Storageに配置したデータからナレッジベースを構築し、それに対してAgentic Retrievalを実行します。
Reasoning Effortを変えたときに質問に対して回答がどれだけリッチになるのかを比較します。

次のように設定したパラメータで検索を実行し、検索クエリと回答を確認します。

1. OutputMode=EXTRACTIVE_DATA / Reasoning Effort=minimal
2. OutputMode=ANSWER_SYNTHESIS / Reasoning Effort=low
3. OutputMode=ANSWER_SYNTHESIS / Reasoning Effort=medium

今回は以下の手順で検証を行います。

1. リソース作成
2. データの用意
3. ナレッジベースの作成
4. Pythonの用意
5. 検証

リソース作成

まず以下のAzureリソースを、AzureポータルやIaCで作成します。

1. Azure AI Search
2. Azure Blob Storage（データ格納用コンテナ）
3. Azure OpenAI / Foundry Models（Chatモデル、Embeddingモデル）

Agentic Retrieval使用においては、以下のモデルが現状サポートされています。

• gpt-4o
• gpt-4o-mini
• gpt-4.1
• gpt-4.1-nano
• gpt-4.1-mini
• gpt-5
• gpt-5-nano
• gpt-5-mini

Azure AI Searchインデックスの作成

まず、ナレッジベースに必要なインデックスを作成します。
今回は以下の手順で作成しますが、本題ではないため、入れるデータの内容と定義を記載するのみとします。

1. Blob Storage上にファイルをアップロード
2. インデックス・インデクサーを作成
3. インデクサーを起動

データとしてIPAが公開している以下のドキュメントを使用します。このデータをBlob Storageに配置します。 www.ipa.go.jp

インデクサーを手動実行し、上記データから検索インデックスを作成します。
ここで使っているのは Azure AI Search の Blob indexerで、PDF、Office（Word / Excel / PowerPoint）、CSVなどの主要形式からテキストを抽出してインデックス化できます。

インデクサー定義

作成されたインデックスの例

ナレッジベースの作成

作成したインデックスからナレッジソースとナレッジベース を作成します。
ナレッジソースは実際に検索をかけるデータそのもの、ナレッジベースはそれらナレッジソースを関連付けて検索・回答に使うまとまりの単位です。

Agentic Retrievalを使用できるよう、ナレッジソースには埋め込みモデルを、ナレッジベースには推論モデルを設定します。

ナレッジソースの定義

ナレッジベースの定義

Pythonの用意

以下のパッケージを含むPython環境を用意します。

• azure-search-documents==11.7.0b2

Agentic Retrievalを利用するには、api-version=2025-11-01-preview に対応したプレビュー版SDKを使用します。

3.2 検証

OutputMode=EXTRACTIVE_DATA / Reasoning Effort=minimal

minimal は LLM によるクエリ計画と回答生成を使わないため、OutputMode=ANSWER_SYNTHESIS は利用できず、EXTRACTIVE_DATA のみ利用可能です。

実際に生成された検索クエリ

1. データ利活用に必要な環境整備についてまとめて

回答として参照した content 抽出例

本ガイドについて
近年、社会を取り巻く環境は目まぐるしく変化しており...
多くの国では、データは国の豊かさや国際競争力の基盤であると捉え...

EXTRACTIVE_DATA は、参照した箇所のテキストを返すだけで、エージェントがそのまま返せる最終回答文は生成しません。

OutputMode=ANSWER_SYNTHESIS / Reasoning Effort=low

実行した結果、下の3つのクエリがエージェントによって作成されました。

1. データ利活用に必要な環境整備 要素一覧（ガバナンス、組織体制、技術基盤、セキュリティ、プライバシー、データ品質、法令遵守、運用プロセス）
2. データガバナンス ベストプラクティス 企業における実装手順
3. データ利活用に必要な技術基盤（データプラットフォーム、データレイク/ウェアハウス、ETL、カタログ、アクセス制御）

回答として返った content 全文

以下は「データ利活用に必要な環境整備」の要点まとめ（箇条書き）。各項目に目的、主要施策、導入の優先度／手順、参考指標・指針を含めます。出典を明記します。

- ガバナンス
  - 目的：データの価値最大化と統制を確保する。[ref_id:1]
  - 主要施策：方針・ルール（収集、保存、アクセス、利用、共有）を定義し統制する。データ品質基準やセキュリティ方針を実施する。[ref_id:0][ref_id:1]
  - 優先度／手順：戦略策定→現状評価→ガバナンス構造設計→ポリシー策定→実施・モニタリング。[ref_id:0]
  - 指標／指針：ISO/IEC38505-1、DMBOKなどの枠組みを参照。ガバナンス原則（責任、戦略等）を指標化。[ref_id:1]

- 組織体制
  - 目的：役割分担と意思決定を明確にし、横断的推進を行う。[ref_id:1]
  - 主要施策：CDOやデータスチュワード、データアーキテクト、データエンジニア等の配置、ガバナンス委員会の設置。[ref_id:1]
  - 優先度／手順：体制設計→必要人材定義→教育・連携体制構築。中央集権型／分散型いずれも戦略共有を確保。[ref_id:1]
  - 指標／指針：役割ごとの責任範囲と運用ルールの可視化（職務定義）。[ref_id:1]

- 技術基盤
  - 目的：データの収集・保存・連携・可視化を効率的に行う。[ref_id:2]
  - 主要施策：データ連携基盤、API、データスペース、データ取引所、ナレッジグラフなどの導入。データカタログやDataOpsの整備。[ref_id:2]
  - 優先度／手順：要件定義→アーキテクチャ設計→既存ビルディングブロック活用→導入・運用。データモデル設計とマスターデータ整備を同時推進。[ref_id:2][ref_id:0]
  - 指標／指針：データ連携ログ、エラーログ解析、DataOps指標、データマチュリティ評価。[ref_id:2]

- セキュリティ／プライバシー
  - 目的：データ漏洩や不正利用を防ぎ、被害最小化を図る。[ref_id:0][ref_id:2]
  - 主要施策：アクセス権限管理、セキュリティポリシー、電子透かし・複製防止技術、APIでのアクセス制御、バックアップ・BCP整備。[ref_id:0][ref_id:2]
  - 優先度／手順：リスク評価→セキュリティ設計（認証・ログ・秘匿等）→実装→定期モニタリング（ランサムウェア対策含む）。[ref_id:2][ref_id:1]
  - 指標／指針：アクセスログ、侵害検知件数、BCPの整備状況。[ref_id:0][ref_id:1]

- データ品質
  - 目的：信頼できるデータを安定的に供給し利活用に耐える品質を保つ。[ref_id:0][ref_id:2]
  - 主要施策：データ品質基準設定、バリデータ導入、データカタログ登録、マスターデータ整備。[ref_id:0][ref_id:2]
  - 優先度／手順：品質基準策定→データ設計とバリデーション導入→定期モニタリングと改善。DataOpsで継続的改善。[ref_id:0][ref_id:2]
  - 指標／指針：データエラー率、品質改善頻度、DataOpsのモニター結果。[ref_id:2]

- 法令／コンプライアンス
  - 目的：法規制遵守とリスク低減を確保する。[ref_id:0][ref_id:1]
  - 主要施策：法令に沿った手続き、セキュリティ・プライバシー方針の実装、アクセス権管理。国際的枠組みへの対応検討。[ref_id:0][ref_id:1]
  - 優先度／手順：関連法令の確認→ポリシー反映→監査・モニタリング。国際連携時はデータ主権の考慮。[ref_id:0]
  - 指標／指針：コンプライアンスチェックリスト、監査結果。参照枠組み：国連提案等（不明な点は詳細確認）。[ref_id:1]

- 運用プロセス
  - 目的：データライフサイクルを通じた安定運用と継続的改善を行う。[ref_id:0]
  - 主要施策：収集→蓄積→設計→整備→運用→活用→廃棄のプロセス定義、モニタリング・ダッシュボード導入。BCPも策定。[ref_id:2][ref_id:0]
  - 優先度／手順：プロセス設計→ツール選定→自動モニタリング整備→定期評価と改善。[ref_id:0]
  - 指標／指針：運用フロー遵守率、モニタリング指標、自動化率。[ref_id:0][ref_id:2]

- スキル・教育
  - 目的：担当者の能力確保と組織内でのデータ活用文化を醸成する。[ref_id:1][ref_id:2]
  - 主要施策：CDOやデータ職種向けの教育、トレーニングプログラム、部門間連携教育。データガバナンスの理解浸透。[ref_id:1]
  - 優先度／手順：必要スキル定義→研修計画→実施→成果モニタリング。現場フィードバックを取り込む。[ref_id:1][ref_id:2]
  - 指標／指針：研修受講率、スキル評価、業務改善事例数。[ref_id:1]

- 導入のロードマップ
  - 目的：段階的かつ確実な導入でリスクを低減し効果を最大化する。[ref_id:0]
  - 主要施策：目標設定→現状評価→設計→段階的実装（優先領域から）→モニタリング→継続的改善。ツール選定とPoC実施を推奨。[ref_id:0]
  - 優先度／手順：まず目的とKPI設定→重要データ領域とユースケース選定→基盤・セキュリティ整備→スキル育成→拡張。[ref_id:0][ref_id:2]
  - 指標／指針：KPI（データマチュリティ、利用率、ROI）、PoC成功基準、モニタリング結果。[ref_id:2][ref_id:0]

出典：各項目の内容は文中の説明に基づく [ref_id:0][ref_id:1][ref_id:2]。

不明：国連提案などの一部枠組みの具体的実装手順や最新ガイドラインの詳細は文中に明確な実施手順が記載されておらず、不明。 [ref_id:1]

low ではクエリ数が増え、回答はそのままエージェントの返答に使える形式になります。
情報量は増えますが、補足が広く、やや冗長に見える部分もありました。

また、回答文中の [ref_id:n] と検索結果側の references の対応関係は AI Search の retrieve レスポンスに含まれており、この参照紐付けをプロンプトで指示しなくても使うことができます。

OutputMode=ANSWER_SYNTHESIS / Reasoning Effort=medium

実行した結果、下の10個のクエリがエージェントによって作成されました。

1. データ利活用に必要な環境整備 全体像 ガイドライン ベストプラクティス
2. データガバナンスとデータ利活用の関係 施策 必要要素
3. データ基盤（データプラットフォーム）に必要な構成要素（データ収集、保存、処理、分析、カタログ、ETL/ELT）
4. セキュリティ・プライバシー対策 データ利活用 必要な措置（アクセス管理、暗号化、ログ監査、個人情報保護）
5. 組織体制と人材育成 データ利活用のための役割（データオーナー、データステュワード、データサイエンティスト）
6. データ利活用 推奨技術スタック データレイク データウェアハウス ETL メタデータ管理 ツール例
7. データガバナンス KPI データ品質指標 定量指標 利用状況 指標例
8. データ利活用 導入ロードマップ 短期 中期 長期 活動項目 工数 事例
9. 中小企業 データガバナンス 実装パターン 低コスト 事例
10. 個人情報保護法 GDPR データ利活用 対応ポイント 要件と対策

回答として返った content 全文

上位概要（要約）
- データ利活用には、技術基盤、組織・ガバナンス、セキュリティ、運用・プロセス、人材の整備が不可欠であり、これらを一体的に設計・運用することが重要である。[ref_id:0][ref_id:2]

詳細チェックリスト（実施項目）
- 技術的要素
  - データ収集・保存・連携基盤の整備：マスタデータ整備やデータの取得・保存・廃棄プロセスを明確化し、相互運用性を確保する（国際標準や政府の相互運用性フレームワークを参照）[ref_id:0][ref_id:1].
  - ツール選定と自動化：データ連携ツール、データベース、メタデータ管理ツール、データ品質管理ツール等を導入し、可能な限りAPI等で自動収集する。[ref_id:0][ref_id:1]
  - 匿名加工・ローカル処理：個人情報は匿名加工や当該国サーバー内での分析等で移転リスクを低減する対応を検討する。[ref_id:3]

- 組織・ガバナンス
  - 体制整備：CDO等の責任者とガバナンス委員会、データマネージャー、データスチュワードを設置する。[ref_id:1]
  - ポリシー策定：収集・保存・アクセス・利用・共有に関する具体的ガイドラインと手続きを定義する（データ戦略に位置付ける）。[ref_id:0][ref_id:1]
  - データ主権と連携ルール：組織横断・境界越え連携時にデータオーナーが利活用を決定できるルールを整備する。[ref_id:0]

- セキュリティ
  - 情報漏洩対策：複製防止技術、電子透かし、アクセス制限できるAPI等を検討・導入する。[ref_id:0][ref_id:2]
  - プライバシー法令順守：各国・地域のデータ移転規定（GDPR、CCPA等）を確認し、同意取得や選択肢提供を行う。[ref_id:3]
  - モニタリングと証跡：変更履歴や操作履歴を可視化し、トレーサビリティを確保する。[ref_id:2]

- 運用・プロセス
  - ライフサイクル管理：入力、変換、保管、取得、利用、廃棄までのフローを定義し統制する。[ref_id:0]
  - モニタリングと評価：ダッシュボード等で自動モニタリングし、定期的に品質評価を行う（DataOpsのフィードバック等）。[ref_id:0][ref_id:2]
  - 市場連携ルール：データ提供時はデータ作成方法や項目説明を規約で公開し、受給者は検索・取得できる仕組みを整備する。[ref_id:3]

- 人材
  - 役割と育成：データスチュワード、データマネージャー等の役割定義と教育・トレーニングを実施する。[ref_id:0][ref_id:1]
  - データリテラシー向上：個人・組織のデータ活用力（データリテラシー・データマチュリティ）を高める施策を行う。[ref_id:1]

実施ポイント（短い補足）
- 可視化を徹底する：ドキュメンテーションやモデリングで共通理解を作ることが基本。[ref_id:0]
- 自動化で現場負担を減らす：手作業を減らし、運用での自動化を進めることが重要。[ref_id:1][ref_id:2]
- 法規制対応を先に設計する：国・地域ごとの規定に合わせて匿名化や同意管理、サーバー運用を決める。[ref_id:3]
- 市場でのデータ流通設計：データ提供者、受給者、運営者の役割と契約・規約を明確にすることが今後の課題。[ref_id:3]

根拠（出典の所在）
- データガバナンスの概念、実施手順や要素（戦略、ルール、データ、ツール、業務プロセス、人材）[ref_id:0].
- データ経営実現のための体制（CDO等）、データ設計・整備、利活用の流れ[ref_id:1].
- セキュリティ・プライバシー対応、国際的なデータ移転やオンライン識別子（Cookie）に関する留意点、およびデータ流通市場の考え方[ref_id:3].

不明な点
- ユーザーの業種や規模、既存の技術スタックが不明なため、具体的なツール選定や実装手順は提示できない（分からない）。

ご要望があれば、業種や現状（既存体制、データ量、法域）を教えてください。該当情報があれば、より具体的なチェックリストや優先順位を提示できます。 [ref_id:0][ref_id:1][ref_id:2][ref_id:3]

medium ではクエリ数がさらに増え、より広範囲を検索する挙動になりました。
一方で回答文は low より短くなり、必要な情報に絞って整理されているように見えました。

実行時間とトークン消費の比較は以下です。
Reasoning Effort=lowとminimalでは、ドキュメントの数に比例して推論トークン数は大きくなりますが、実行時間は大きくは増えないという結果でした。

条件	実行時間(秒)	入力トークン数	出力トークン数	推論トークン数
OutputMode=EXTRACTIVE_DATA / Reasoning Effort=minimal	2.73	0	0	64,739
OutputMode=ANSWER_SYNTHESIS / Reasoning Effort=low	38.85	10,491	3,148	194,988
OutputMode=ANSWER_SYNTHESIS / Reasoning Effort=medium	47.52	17,265	3,423	650,712

検証に使った Python コード

#!/usr/bin/env python3

import argparse
import json
import os
import time
from pathlib import Path
from typing import Any

from azure.core.credentials import AzureKeyCredential
from azure.search.documents.knowledgebases import KnowledgeBaseRetrievalClient
from azure.search.documents.knowledgebases.models import (
    KnowledgeBaseMessage,
    KnowledgeBaseMessageTextContent,
    KnowledgeBaseRetrievalRequest,
    KnowledgeRetrievalLowReasoningEffort,
    KnowledgeRetrievalMediumReasoningEffort,
    KnowledgeRetrievalMinimalReasoningEffort,
    KnowledgeRetrievalOutputMode,
)
from dotenv import load_dotenv

SCRIPT_DIR = Path(__file__).resolve().parent
DEFAULT_QUESTION = "データ利活用に必要な環境整備についてまとめて"
DEFAULT_OUTPUT_MODE = "answer_synthesis"
DEFAULT_REASONING_EFFORT = "low"
SYSTEM_PROMPT = (
    "あなたは有能なアシスタントです。"
    "ナレッジベースを使って回答してください。"
    "分からない場合は分からないと答えてください。"
)

load_dotenv(SCRIPT_DIR / ".env", override=False)


def require_env(name: str) -> str:
    value = os.environ.get(name, "").strip()
    if not value:
        raise RuntimeError(f"Environment variable {name} is required.")
    return value


def build_client() -> KnowledgeBaseRetrievalClient:
    return KnowledgeBaseRetrievalClient(
        endpoint=require_env("SEARCH_ENDPOINT").rstrip("/"),
        knowledge_base_name=os.environ.get("KNOWLEDGE_BASE_NAME", "sample-knowledge"),
        credential=AzureKeyCredential(require_env("SEARCH_API_KEY")),
    )


def build_reasoning_effort(kind: str) -> object:
    if kind == "minimal":
        return KnowledgeRetrievalMinimalReasoningEffort()
    if kind == "low":
        return KnowledgeRetrievalLowReasoningEffort()
    return KnowledgeRetrievalMediumReasoningEffort()


def build_request(
    question: str, output_mode: str, reasoning_effort: str
) -> KnowledgeBaseRetrievalRequest:
    if output_mode == "extractive_data":
        # minimal + EXTRACTIVE_DATA は intents で直接検索
        if reasoning_effort == "minimal":
            return KnowledgeBaseRetrievalRequest(
                intents=[{"type": "semantic", "search": question}],
                retrieval_reasoning_effort=build_reasoning_effort(reasoning_effort),
                output_mode=KnowledgeRetrievalOutputMode.EXTRACTIVE_DATA,
                include_activity=True,
            )

        # low / medium + EXTRACTIVE_DATA は message 経由でクエリ計画
        return KnowledgeBaseRetrievalRequest(
            messages=[
                KnowledgeBaseMessage(
                    role="user",
                    content=[KnowledgeBaseMessageTextContent(text=question)],
                )
            ],
            retrieval_reasoning_effort=build_reasoning_effort(reasoning_effort),
            output_mode=KnowledgeRetrievalOutputMode.EXTRACTIVE_DATA,
            include_activity=True,
        )

    # ANSWER_SYNTHESIS は low / medium のみ
    return KnowledgeBaseRetrievalRequest(
        messages=[
            KnowledgeBaseMessage(
                role="system",
                content=[KnowledgeBaseMessageTextContent(text=SYSTEM_PROMPT)],
            ),
            KnowledgeBaseMessage(
                role="user",
                content=[KnowledgeBaseMessageTextContent(text=question)],
            ),
        ],
        retrieval_reasoning_effort=build_reasoning_effort(reasoning_effort),
        output_mode=KnowledgeRetrievalOutputMode.ANSWER_SYNTHESIS,
        include_activity=True,
        max_runtime_in_seconds=60,
        # API 制約: max_output_size は 5000 より大きい値が必要
        max_output_size=6000,
    )


def extract_query_count(activity: Any) -> int:
    if not isinstance(activity, list):
        return 0

    count = 0
    for record in activity:
        if not isinstance(record, dict):
            continue
        for key in (
            "azureBlobArguments",
            "azure_blob_arguments",
            "searchIndexArguments",
            "search_index_arguments",
        ):
            args_value = record.get(key)
            if isinstance(args_value, dict) and isinstance(args_value.get("search"), str):
                count += 1
                break
    return count


def extract_queries(activity: Any) -> list[str]:
    if not isinstance(activity, list):
        return []

    queries: list[str] = []
    for record in activity:
        if not isinstance(record, dict):
            continue
        for key in (
            "azureBlobArguments",
            "azure_blob_arguments",
            "searchIndexArguments",
            "search_index_arguments",
        ):
            args_value = record.get(key)
            if isinstance(args_value, dict) and isinstance(args_value.get("search"), str):
                queries.append(args_value["search"])
                break
    return queries


def extract_content_text(payload: dict[str, Any]) -> str:
    # response.content[*].text を連結して表示用 text を作る
    response = payload.get("response")
    if not isinstance(response, list):
        return ""

    texts: list[str] = []
    for message in response:
        if not isinstance(message, dict):
            continue
        content = message.get("content")
        if not isinstance(content, list):
            continue
        for item in content:
            if isinstance(item, dict) and isinstance(item.get("text"), str):
                texts.append(item["text"])
    return "\n".join(texts).strip()


def to_int(value: Any) -> int:
    try:
        return int(value)
    except (TypeError, ValueError):
        return 0


def extract_activity_metrics(activity: Any) -> dict[str, Any]:
    metrics = {
        "query_planning_input_tokens": 0,
        "query_planning_output_tokens": 0,
        "answer_synthesis_input_tokens": 0,
        "answer_synthesis_output_tokens": 0,
        "reasoning_tokens": 0,
        "activity_elapsed_total_ms": 0,
        "activity_elapsed_breakdown_ms": {},
    }
    if not isinstance(activity, list):
        return metrics

    for record in activity:
        if not isinstance(record, dict):
            continue

        record_type = str(record.get("type", ""))
        elapsed_ms = to_int(record.get("elapsed_ms"))
        if elapsed_ms > 0:
            metrics["activity_elapsed_total_ms"] += elapsed_ms
            breakdown = metrics["activity_elapsed_breakdown_ms"]
            breakdown[record_type] = to_int(breakdown.get(record_type)) + elapsed_ms

        if record_type == "modelQueryPlanning":
            metrics["query_planning_input_tokens"] += to_int(record.get("input_tokens"))
            metrics["query_planning_output_tokens"] += to_int(record.get("output_tokens"))
        elif record_type == "modelAnswerSynthesis":
            metrics["answer_synthesis_input_tokens"] += to_int(record.get("input_tokens"))
            metrics["answer_synthesis_output_tokens"] += to_int(record.get("output_tokens"))
        elif record_type == "agenticReasoning":
            metrics["reasoning_tokens"] += to_int(record.get("reasoning_tokens"))

    return metrics


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Run Agentic Retrieval with direct output_mode / reasoning_effort."
    )
    parser.add_argument("--question", default=DEFAULT_QUESTION)
    parser.add_argument(
        "--output-mode",
        choices=["extractive_data", "answer_synthesis"],
        default=DEFAULT_OUTPUT_MODE,
    )
    parser.add_argument(
        "--reasoning-effort",
        choices=["minimal", "low", "medium"],
        default=DEFAULT_REASONING_EFFORT,
    )
    parser.add_argument(
        "--print-raw",
        action="store_true",
        help="デバッグ用に raw_response_json を表示する",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    if (
        args.output_mode == "answer_synthesis"
        and args.reasoning_effort == "minimal"
    ):
        raise RuntimeError(
            "minimal の場合は output_mode=extractive_data のみ指定できます。"
        )

    client = build_client()
    start = time.perf_counter()
    payload = client.retrieve(
        build_request(args.question, args.output_mode, args.reasoning_effort)
    ).as_dict()
    end_to_end_elapsed_ms = int((time.perf_counter() - start) * 1000)

    content_text = extract_content_text(payload)
    activity = payload.get("activity")
    queries = extract_queries(activity)
    activity_metrics = extract_activity_metrics(activity)
    input_tokens_total = (
        activity_metrics["query_planning_input_tokens"]
        + activity_metrics["answer_synthesis_input_tokens"]
    )
    output_tokens_total = (
        activity_metrics["query_planning_output_tokens"]
        + activity_metrics["answer_synthesis_output_tokens"]
    )
    summary = {
        "output_mode": args.output_mode,
        "reasoning_effort": args.reasoning_effort,
        "reference_count": len(payload.get("references", []))
        if isinstance(payload.get("references"), list)
        else 0,
        "query_count": extract_query_count(activity),
        "queries": queries,
        "answer_char_count": len(content_text),
        "timing": {
            "end_to_end_elapsed_ms": end_to_end_elapsed_ms,
            "activity_elapsed_total_ms": activity_metrics["activity_elapsed_total_ms"],
            "activity_elapsed_breakdown_ms": activity_metrics[
                "activity_elapsed_breakdown_ms"
            ],
        },
        "token_usage": {
            "query_planning": {
                "input_tokens": activity_metrics["query_planning_input_tokens"],
                "output_tokens": activity_metrics["query_planning_output_tokens"],
            },
            "answer_synthesis": {
                "input_tokens": activity_metrics["answer_synthesis_input_tokens"],
                "output_tokens": activity_metrics["answer_synthesis_output_tokens"],
            },
            "reasoning_tokens": activity_metrics["reasoning_tokens"],
            "input_tokens_total": input_tokens_total,
            "output_tokens_total": output_tokens_total,
        },
    }
    print(json.dumps(summary, ensure_ascii=False, indent=2))

    if content_text:
        print("content_text:")
        print(content_text)

    if args.print_raw:
        print("raw_response_json:")
        print(json.dumps(payload, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

4. まとめ

今回はAzure AI SearchのAgentic Retrievalを使ってエージェント検索を簡単に実現しました。
ナレッジベースを用意するだけで簡単に自律的に検索して情報を返してくれるエージェントが作成できるのは便利ですね。
Reasoning Effortを設定することでより詳細な情報を回答に含められるなど操作できるのもうれしいポイントです。

Acroquest Technologyでは、キャリア採用を行っています。

• Azure OpenAI／Amazon Bedrock等を使った生成AIソリューションの開発
• ディープラーニング等を使った自然言語／画像／音声／動画解析の研究開発
• マイクロサービス、DevOps、最新のOSSやクラウドサービスを利用する開発プロジェクト
• 書籍・雑誌等の執筆や、社内外での技術の発信・共有によるエンジニアとしての成長

少しでも上記に興味を持たれた方は、是非以下のページをご覧ください。 www.wantedly.com