Amazon Q Business で Agentic RAG を構築する ― データソース4パターンの選定と boto3 実装リファレンス

はじめに

Amazon Q Business は、企業内データソースに対してRAGベースの自律的なQ&Aを提供するAWS マネージドAI Agentです。単なる検索エンジンではなく、複数データソースを横断して回答を生成・引用まで行う Agentic AI として、社内ナレッジ活用・カスタマーサポート・業務自動化の文脈で採用が増えています。

本記事は実際のboto3 API検証と設計パターンの整理をもとに、提案・デリバリーで繰り返し参照できるリファレンスとしてまとめたものです。「どのデータソースをどのパターンで繋ぐか」「アクセス制御をどう設計するか」の判断軸を体系化します。

Amazon Q Business のアーキテクチャ全体像

Amazon Q Business は以下の4層で構成されます。

【データソース層】
  S3（ドキュメント）   Athena（構造化）   Salesforce / SharePoint   カスタムAPI
        |                   |                      |                      |
        +-------------------+----------------------+----------------------+
                                     |
                            【インジェスト層】
                     ネイティブコネクタ / カスタムコネクタ
                      （スケジュール同期 or オンデマンド）
                                     |
                            【インデックス層】
                              Amazon Q Index
                         チャンク化 / 埋め込み / ACL付与
                                     |
                          【アプリケーション層】
              +------------------+------------------+
              |                                     |
       Web Experience                         ChatSync API
       （チャット UI）                     （プログラム呼び出し）

各層の役割：

Amazon Q Business が Agentic AI である理由: ユーザーの質問に対して、Agentが自律的に複数ソースを横断検索 → 関連チャンクを選択 → 回答を合成 → 引用を付与、という多段階の判断を人手を介さず実行します。

Agentic RAG の動作シーケンス

ユーザー: 「Masks の在庫が発注点を下回っている。過去の緊急発注手順を教えて」
       |
       | ChatSync API
       v
  Amazon Q Agent
       |
       +-- [1] クエリ解析: 「在庫不足 + 緊急発注手順」と判断
       |
       +-- [2] インデックス横断検索（自律）
       |         S3（調達マニュアル.pdf）  ← 「緊急発注手順」にヒット
       |         Athena（在庫履歴テーブル）← 「Masks の過去発注履歴」にヒット
       |
       +-- [3] 関連チャンクのスコアリング・選択（自律）
       |         上位チャンクを選択し、回答合成に渡す
       |
       +-- [4] 回答合成（自律）
       |         「手順書p.3に基づき、緊急発注はXXX。Masksの直近発注は〇月〇日」
       |
       +-- [5] 引用付与（自律）
                 sourceAttributions: [調達マニュアル.pdf#p3, 在庫履歴#row-47]

このシーケンスはユーザーが1回質問するだけで Agent が自律的に実行します。以下は ChatSync API を使って実際にこの動作を確認した例です。

import boto3

qb = boto3.client('qbusiness', region_name='us-east-1')

response = qb.chat_sync(
    applicationId='app-xxxxxxxx',
    userId='user@example.com',
    userMessage='Masksの在庫が発注点を下回っています。過去の緊急発注手順を教えてください'
)

print(response['systemMessage'])
# → "調達マニュアルの手順に基づき、緊急発注は購買部門へのメール申請が必要です（マニュアル p.3）。
#    Masksの直近発注は2026年2月14日で、200個を発注済みです（在庫履歴より）。"

for source in response.get('sourceAttributions', []):
    print(f"引用: {source['title']} — {source['url']}")
# → 引用: 調達マニュアル.pdf — s3://company-docs/procurement/manual.pdf
# → 引用: 在庫発注履歴 — athena://inventory-db/order_history#row-47

レスポンス例（sourceAttributions 抜粋）:

{
"systemMessage": "調達マニュアルの手順に基づき、緊急発注は購買部門へのメール申請が必要です（マニュアル p.3）。Masksの直近発注は2026年2月14日で、200個を発注済みです（在庫履歴より）。",
"sourceAttributions": [
{
"title": "調達マニュアル.pdf",
"url": "s3://company-docs/procurement/manual.pdf",
"citationNumber": 1,
"snippet": "緊急発注が必要な場合は、購買部門（purchase@example.com）へ件名「緊急発注申請」でメール申請…"
},
{
"title": "在庫発注履歴",
"url": "athena://inventory-db/order_history",
"citationNumber": 2,
"snippet": "2026-02-14 | Masks | 200個 | 発注済み | 担当: 田中"
}
]
}

sourceAttributions に複数ソース（S3文書＋Athenaクエリ結果）が含まれていることが、Agent が複数データソースを横断して自律的に回答生成した証拠です。

データソース分類と選定フレームワーク

Amazon Q Business のデータソースは3パターンに分類できます。

データ特性
    |
    +-- 非構造化ドキュメント系 --> パターン1: ネイティブコネクタ（S3 / SharePoint / Confluence）
    |
    +-- 構造化DB / DWH系      --> パターン2: Athena経由（S3 + Glue Catalog）
    |
    +-- SaaS / CRM / ITSM系   --> パターン3: SaaS ネイティブコネクタ（Salesforce / ServiceNow）
    |
    +-- 既存システム / API系   --> パターン4: カスタムコネクタ（BatchPutDocument API）

選定マトリクス

パターン1 — S3 ネイティブコネクタ

構成

社内文書（PDF / Word / HTML）
        |
        v アップロード
    S3 Bucket
   /docs/ /manuals/
        |
        | スケジュール同期 or オンデマンド
        v
  S3 Data Source Connector  <-- metadata/ (属性JSON)
        |
        v チャンク化 / 埋め込み
    Amazon Q Index

boto3 実装

import boto3

qb = boto3.client('qbusiness', region_name='us-east-1')

response = qb.create_data_source(
    applicationId='app-xxxxxxxx',
    indexId='idx-xxxxxxxx',
    displayName='社内文書-S3',
    type='S3',
    configuration={
        'S3Configuration': {
            'bucketName': 'company-knowledge-bucket',
            'inclusionPrefixes': ['docs/', 'manuals/'],
            'exclusionPatterns': ['*.tmp', '*/archive/*'],
            'documentsMetadataConfiguration': {
                'S3Prefix': 'metadata/'
            }
        }
    },
    syncSchedule='cron(0 2 * * ? *)',  # 毎日 02:00 UTC
    roleArn='arn:aws:iam::123456789012:role/QBusinessDataSourceRole'
)

data_source_id = response['dataSourceId']

# 手動同期
qb.start_data_source_sync_job(
    applicationId='app-xxxxxxxx',
    indexId='idx-xxxxxxxx',
    dataSourceId=data_source_id
)

メタデータスキーマ設計

S3オブジェクトと同名の .metadata.json をメタデータプレフィックス下に置くことで属性フィルタリングが可能になります。

{
"_source_uri": "https://s3.amazonaws.com/bucket/docs/hr-policy.pdf",
"_category": "HR",
"_last_updated_at": "2026-04-01T00:00:00Z",
"department": "人事部",
"confidentiality": "internal"
}

設計上の注意点:
– _ プレフィックスはシステム予約属性。カスタム属性はプレフィックスなしで定義する
– ファイルソース系データセットは describe_data_set 系APIで列情報を取得できない（File source type is not supported in Public API）。属性確認は get_index または list_data_sources 経由で行う

パターン2 — Athena 経由の構造化データ

構成

RDS / Redshift
（構造化データ）
       |
       | ETL / Export
       v
  S3 (CSV / Parquet)
       |
       v
  Glue Catalog
（スキーマ定義）
       |
       v
    Athena
  （SQL変換クエリ）
       |
       | クエリ結果をMarkdown / JSONドキュメント化
       v
  S3 (docs/)
       |
       v
  S3 Connector --> Q Index

構造化データをRAGに乗せる際の主な課題は「テーブルの行をどう文書化するか」です。単純なCSVダンプではなく、行ごとに文脈を持つドキュメントに変換する必要があります。

import boto3

athena = boto3.client('athena', region_name='ap-northeast-1')
s3 = boto3.client('s3')

def export_as_documents(query: str, output_bucket: str, output_prefix: str):
    response = athena.start_query_execution(
        QueryString=query,
        ResultConfiguration={
            'OutputLocation': f's3://{output_bucket}/athena-results/'
        }
    )

    # ポーリングで完了待ち（省略）

    results = athena.get_query_results(
        QueryExecutionId=response['QueryExecutionId']
    )

    columns = [col['Label'] for col in results['ResultSet']['ResultSetMetadata']['ColumnInfo']]

    for i, row in enumerate(results['ResultSet']['Rows'][1:]):
        values = {col: row['Data'][j].get('VarCharValue', '') for j, col in enumerate(columns)}

        # 行をMarkdown形式のドキュメントに整形
        doc_content = '\n'.join([f'**{k}**: {v}' for k, v in values.items()])

        s3.put_object(
            Bucket=output_bucket,
            Key=f'{output_prefix}/row-{i}.md',
            Body=doc_content.encode('utf-8')
        )

注意: 大量レコードを全行ドキュメント化するとインデックスコストが増大します。集約・サマリー化するか、カスタムコネクタ（パターン4）でフィルタ条件を付けて件数を絞るアプローチが現実的です。

パターン3 — SaaS ネイティブコネクタ（Salesforce）

SaaS側の権限（ACL）をそのままQ Businessに引き継げる点が最大のメリットです。

構成

Salesforce                    IAM Identity Center
（Account / Case / KB）        （SAML連携）
       |                              |
       | OAuth 2.0 認証               | ユーザーコンテキスト
       v                              v
  Salesforce Connector --> Q Index --> Q Application
                                           |
                                           v
                              ユーザーが閲覧可能な
                              ドキュメントのみ回答

Secrets Manager でクレデンシャル管理

import boto3, json

sm = boto3.client('secretsmanager', region_name='us-east-1')
qb = boto3.client('qbusiness', region_name='us-east-1')

secret_arn = sm.create_secret(
    Name='qbusiness/salesforce-credentials',
    SecretString=json.dumps({
        'clientId': 'YOUR_SF_CLIENT_ID',
        'clientSecret': 'YOUR_SF_CLIENT_SECRET',
        'username': 'admin@example.com',
        'password': 'PASSWORD',
        'securityToken': 'SECURITY_TOKEN'
    })
)['ARN']

qb.create_data_source(
    applicationId='app-xxxxxxxx',
    indexId='idx-xxxxxxxx',
    displayName='Salesforce-KB',
    type='SALESFORCE',
    configuration={
        'SalesforceConfiguration': {
            'serverUrl': 'https://yourinstance.salesforce.com',
            'secretArn': secret_arn,
            'crawlKnowledgeArticles': True,
            'crawlAttachments': False,
            'standardObjectConfigurations': [
                {
                    'name': 'KNOWLEDGE_ARTICLE',
                    'query': "ArticleType = 'FAQ__kav'",
                    'fieldMappings': [
                        {
                            'dataSourceFieldName': 'Title',
                            'indexFieldName': '_document_title',
                            'indexFieldType': 'STRING'
                        }
                    ]
                }
            ]
        }
    },
    roleArn='arn:aws:iam::123456789012:role/QBusinessSalesforceRole'
)

SaaS コネクタ共通の注意点:
– 初回同期は全量取得のため、大規模なSaaSは数時間かかることがある
– SaaS側のAPI Rate Limitに注意（Salesforceは query() の同時呼び出し制限あり）
– ACL引き継ぎはSaaS側の権限変更が次回同期まで反映されない点を考慮する

パターン4 — カスタムコネクタ（BatchPutDocument API）

既存システム連携や独自フォーマットのデータには batch_put_documents APIが最も柔軟です。

処理フロー

基幹システム / API
       |
       | データ取得（差分 / 全量）
       v
  Lambda（変換・整形）
  - テキスト化
  - メタデータ付与
  - ACL設定
       |
       v
  batch_put_documents()
       |
       +-- 成功 --> Q Index に反映
       |
       +-- 失敗 --> 失敗IDをリトライキューへ
                    (SQS / DynamoDB)

実装

import boto3

qb = boto3.client('qbusiness', region_name='us-east-1')

def ingest_documents(application_id: str, index_id: str, documents: list):
    formatted = []
    for doc in documents:
        formatted.append({
            'id': doc['id'],
            'title': doc['title'],
            'content': {
                'blob': doc['content'].encode('utf-8')
            },
            'contentType': 'PLAIN_TEXT',
            'documentAttributes': [
                {'name': '_category',         'value': {'stringValue': doc.get('category', 'general')}},
                {'name': '_last_updated_at',  'value': {'dateValue': doc['updated_at']}},
                {'name': 'source_system',     'value': {'stringValue': doc.get('source', 'unknown')}},
            ],
            'accessConfiguration': {
                'accessControls': [
                    {
                        'principals': [
                            {'user': {'id': uid, 'type': 'USER'}}
                            for uid in doc.get('allowed_users', [])
                        ],
                        'memberRelation': 'OR'
                    }
                ],
                'memberRelation': 'OR'
            }
        })

    # 10件ずつバッチ処理（API上限）
    for i in range(0, len(formatted), 10):
        response = qb.batch_put_documents(
            applicationId=application_id,
            indexId=index_id,
            documents=formatted[i:i+10]
        )
        for failed in response.get('failedDocuments', []):
            print(f"失敗: {failed['id']} — {failed['error']['errorMessage']}")

アクセス制御設計パターン

Amazon Q Business のアクセス制御は2軸で設計します。

軸1: アプリケーションレベル認証

認証方式の選択
    |
    +-- 社員 / パートナーがブラウザから使う
    |       --> IAM Identity Center + SAML連携
    |
    +-- 社内システムからAPIで呼び出す
    |       --> IAM認証
    |
    +-- 外部公開（顧客向け）
            --> Cognito + カスタムUI
                （Q Business Web Experience は使わない）

軸2: ドキュメントレベルACL

提案・デリバリー時の考慮点

コスト設計

同期戦略の選定

更新頻度による分岐
    |
    +-- 月次以下      --> フル同期（週次スケジュール）
    |
    +-- 日次程度      --> 増分同期（_last_updated_at でフィルタ）
    |
    +-- 時間単位      --> オンデマンド同期（EventBridge Trigger + batch_put_documents）
    |
    +-- 分単位        --> ストリーミング（Kinesis + Lambda + batch_put_documents）

増分同期の実装例：

from datetime import datetime, timezone

def incremental_sync(application_id: str, index_id: str, source_api):
    last_sync = get_last_sync_timestamp()  # DynamoDB等から取得

    updated_docs = source_api.get_documents_since(last_sync)
    if updated_docs:
        ingest_documents(application_id, index_id, updated_docs)

    deleted_ids = source_api.get_deleted_since(last_sync)
    if deleted_ids:
        qb.batch_delete_document(
            applicationId=application_id,
            indexId=index_id,
            documentIdList=deleted_ids
        )

    save_last_sync_timestamp(datetime.now(timezone.utc))

インデックス容量設計

STARTERで開始してENTERPRISEにアップグレードする段階的アプローチが提案フェーズでは現実的です。

監視設計

def monitor_sync_jobs(application_id, index_id, data_source_id):
    response = qb.list_data_source_sync_jobs(
        applicationId=application_id,
        indexId=index_id,
        dataSourceId=data_source_id,
        maxResults=5
    )
    for job in response['history']:
        m = job.get('metrics', {})
        print(f"状態: {job['status']} | 追加: {m.get('documentsAdded',0)} | "
              f"更新: {m.get('documentsModified',0)} | 失敗: {m.get('documentsFailed',0)}")

同期失敗率が継続上昇する場合、データソース側のAPI制限・認証トークン期限切れ・ドキュメントサイズ超過のいずれかが原因であることがほとんどです。

落とし穴まとめ

まとめ

Amazon Q Business のデータソース連携は「何を・いつ・誰に見せるか」の3軸で設計パターンが決まります。

Quick Flows を使ったフロー自動化と本記事のデータ連携パターンを組み合わせると、データ取得 → AI判断 → アクション通知 のエンドツーエンドなAgentic AIパイプラインが構成できます。
Amazon Quick Flows によるAI業務フロー自動化