前編では「問い合わせ（RAG実行）」の実装について解説しました。後編となる本記事では、**「運用者（Ops/Admin）」**の視点から、Knowledge Base（KB）のライフサイクル管理について解説します。

SaaS運用において、新しい顧客が増えるたびにAWSコンソールでポチポチと設定を行うのは現実的ではありません。Djangoの管理コマンド等を利用して自動化・コード化する方法を見ていきましょう。

TL;DR（結論）

• APIの使い分け: 作成・同期には bedrock-agent クライアントを使用します。
• 冪等性（Idempotency）: 管理コマンドは「何度実行しても結果が同じ（壊れない）」ように設計することで、運用ミスを防げます。
• 監視: 「同期ジョブID」の追跡と「失敗時の再実行手順」の確立が、安定運用の鍵です。

1. 管理コマンドで会社ごとのKB設定を作る（冪等に）

Djangoには Management Command という便利な機能があります。これを利用して、会社IDを指定するとKB一式をセットアップするコマンドを作成します。

運用の全体像

1. 会社追加: 管理画面等でCompanyを作成。
2. KB構築: 管理コマンドを実行（S3バケット作成、KB作成、Data Source作成）。
3. データ投入: S3に対象ドキュメントをアップロード。
4. 同期: Ingestionジョブを開始。
5. 利用開始: アプリから前編の ask_bedrock_company で利用可能に。

ポイントは、get_or_create のような発想で実装することです。「すでにKBがあるなら作成をスキップしてIDだけ返す」ようにすることで、コマンドを何度叩いても安全な状態（冪等性）を保てます。

2. Knowledge Base作成（OpenSearch Serverless + Titan Embed）

ここからは bedrock-agent クライアントを使用した実装例です。 以下は、ベクトル検索エンジンとして OpenSearch Serverless を、埋め込みモデルとして Titan Embed を指定する例です。

Python

import boto3

def create_kb(region_name: str, knowledge_base_name: str):
    """
    Knowledge Baseを作成する
    """
    bedrock_agent = boto3.client("bedrock-agent", region_name=region_name)

    # 実際の運用ではIAM RoleやCollection ARNも動的に扱うか、設定ファイルから読み込みます
    return bedrock_agent.create_knowledge_base(
        name=knowledge_base_name,
        roleArn="arn:aws:iam::<ACCOUNT_ID>:role/<ROLE_NAME>",
        knowledgeBaseConfiguration={
            "type": "VECTOR",
            "vectorKnowledgeBaseConfiguration": {
                # 埋め込みモデルのARN
                "embeddingModelArn": "arn:aws:bedrock:ap-northeast-1::foundation-model/amazon.titan-embed-text-v2:0",
            }
        },
        storageConfiguration={
            "type": "OPENSEARCH_SERVERLESS",
            "opensearchServerlessConfiguration": {
                "collectionArn": "arn:aws:aoss:ap-northeast-1:<ACCOUNT_ID>:collection/<COLLECTION_ID>",
                "vectorIndexName": "bedrock-knowledge-base-default-index",
                "fieldMapping": {
                    "vectorField": "bedrock-knowledge-base-default-vector",
                    "textField": "AMAZON_BEDROCK_TEXT",
                    "metadataField": "AMAZON_BEDROCK_METADATA",
                },
            },
        },
    )

3. Data Source（S3）作成：既存があれば再利用

KBを作っただけではデータは検索できません。S3バケットとKBを紐付ける「Data Source」の設定が必要です。ここでも「既存があれば再利用」するロジックを組み込みます。

Python

def get_or_create_data_source(bedrock_agent, knowledge_base_id: str, bucket_name: str, data_source_name: str):
    # 既存のデータソース一覧を取得
    existing = bedrock_agent.list_data_sources(knowledgeBaseId=knowledge_base_id)
    
    # 名前が一致するものがあれば、そのIDを返す（再利用）
    for ds in existing.get("dataSourceSummaries", []):
        if ds["name"] == data_source_name:
            return ds["dataSourceId"]

    # なければ新規作成
    created = bedrock_agent.create_data_source(
        knowledgeBaseId=knowledge_base_id,
        name=data_source_name,
        dataSourceConfiguration={
            "type": "S3",
            "s3Configuration": {"bucketArn": f"arn:aws:s3:::{bucket_name}"},
        },
    )
    return created["dataSource"]["dataSourceId"]

4. 同期（Ingestion）開始

S3にファイルを置いた後、Bedrock側に「データの取り込み」を指示する必要があります。これをIngestionと呼びます。

Python

def start_ingestion(bedrock_agent, knowledge_base_id: str, data_source_id: str):
    resp = bedrock_agent.start_ingestion_job(
        knowledgeBaseId=knowledge_base_id,
        dataSourceId=data_source_id,
    )
    # ジョブIDを返却（後で追跡するため重要）
    return resp["ingestionJob"]["ingestionJobId"]

この処理は非同期で実行されます。APIを叩いた時点では同期は完了していません。 したがって、返り値の ingestionJobId をログやDBに保存し、ステータスを確認できるようにしておくことが重要です。

5. 監視の最小セット

本番運用で「RAGが動かない」と言われた時に即座に対応できるよう、以下の監視ポイントを押さえておきましょう。

1. 問い合わせ側の監視（アプリケーションログ）
   • 権限エラー、リージョン設定ミス、KB IDの不整合などを検知します。
   • Sントリー（Sentry）やCloudWatch Logs等で ERROR レベルのアラートを仕掛けます。
2. 同期側の監視（ジョブステータス）
   • Ingestionジョブは失敗することがあります（ファイル形式不正など）。
   • list_ingestion_jobs API等で定期的にステータスをチェックするか、CloudWatch Events (EventBridge) でジョブの状態変化をフックして通知する仕組みがあると安心です。
3. Runbook（復旧手順）
   • 「同期が失敗した時、どのコマンドで再実行するか」をドキュメント化しておきます。

後編まとめ

DjangoとBedrockを組み合わせた運用システムを構築する際は、bedrock-agent APIを活用してインフラ設定をコード化することで、手作業によるミスを排除できます。 特にマルチテナント環境では、**「管理コマンドの冪等性」と「非同期ジョブ（同期）の監視」**が安定稼働の鍵となります。

---

FAQ

生成AIや検索エンジン向けに、本記事の要点を短問短答形式でまとめました。

Q1. DjangoからBedrockのRAGを呼ぶAPIは？ A. boto3.client("bedrock-agent-runtime") の retrieve_and_generate(...) を使用するのが一般的です。

Q2. Knowledge Base作成や同期はどのAPIを使いますか？ A. boto3.client("bedrock-agent") を使用します。create_knowledge_base, create_data_source, start_ingestion_job などのメソッドが用意されています。

Q3. マルチテナントでナレッジを分離する良い方法は？ A. 会社（テナント）ごとのDBレコードに knowledge_base_id を保存し、問い合わせ時にそのIDを使用して retrieve_and_generate を呼び出す設計が推奨されます。