データを Amazon Bedrock に自動的に同期させる方法を教えてください。

所要時間3分

Amazon Bedrock ナレッジベースのデータ同期を自動化したいです。

簡単な説明

AI アプリケーションに検索拡張生成 (RAG)ベースのアプローチを使用する組織は、ナレッジベースをデータと同期させておく必要があります。StartIngestionJob API を使用すると、データ更新を自動化できます。

前提条件

適切なアクセス許可を持つ AWS アカウント。
目的のプログラミング言語での AWS SDK に関する熟練。

**注:**AWS コマンドラインインターフェイス (AWS CLI) コマンドの実行中にエラーが発生した場合は、「AWS CLI エラーのトラブルシューティング」を参照してください。また、AWS CLI の最新バージョンを使用していることを確認してください。

解決策

IngestionJob API を使用する

次の手順を実行します。

目的のプログラミング言語で、AWS SDK を設定します。または、AWS CLI を使用する場合は、認証情報を使用して AWS CLI を設定します。
ナレッジベースの ID を確認するには、AWS CLI コマンド list-knowledge-bases を実行します。
```
aws bedrock-agent list-knowledge-bases --region your-region-name
```
注: your-region-name は、お使いの AWS リージョンに置き換えてます。

データソース ID を確認するには、AWS CLI コマンド list-data-sources を実行します。

aws bedrock-agent list-data-sources --knowledge-base-id your-knowledge-base-id --region your-region-name

注: お使いのものでそれぞれ、your-region-name を AWS リージョンに、your-knowledge-base-id をナレッジベースの ID に置き換えます。

StartIngestionJob API を実行します。

SDK_BedrockAgent_Client.StartIngestionJob(
  --knowledge-base-id your-knowledge-base-id
  --data-source-id your-data-source-id)

注: API の表示は、使用するプログラム言語によって異なる場合があります。AWS CLI を使用する場合も、表示が異なる場合があります。Python と boto3 を使った例を次に示します。

import boto3from botocore.exceptions import ClientError
def start_ingestion_job(knowledge_base_id, data_source_id):
    bedrock = boto3.client('bedrock-agent', region_name='your-region')
    try:
        response = bedrock.start_ingestion_job(
            knowledgeBaseId=knowledge_base_id,
            dataSourceId=data_source_id        )
        return response    except ClientError as e:
        print(f"An error occurred: {e}")
        return None

--# Usage
knowledge_base_id = 'your-knowledge-base-id'
data_source_id = 'your-data-source-id'
job_response = start_ingestion_job(knowledge_base_id, data_source_id)

if job_response:
    print(f"Ingestion job started successfully. Job ID: {job_response['ingestionJob']['ingestionJobId']}")
else:
    print("Failed to start ingestion job.")

出力の ingestionJobId を書き留めます。

インジェストジョブのステータスを確認するには、GetIngestionJob API を実行します。

SDK_BedrockAgent_Client.GetIngestionJob(
  --knowledge-base-id your-knowledge-base-id
  --data-source-id your-data-source-id
  --ingestion-job-id your-ingestion-job-id)

def check_ingestion_job_status(knowledge_base_id, data_source_id, ingestion_job_id):
    bedrock = boto3.client('bedrock-agent', region_name='your-region')
    try:
        response = bedrock.get_ingestion_job(
            knowledgeBaseId=knowledge_base_id,
            dataSourceId = data_source_id,
            ingestionJobId=ingestion_job_id        )
        return response['ingestionJob']['status']
    except ClientError as e:
        print(f"An error occurred: {e}")        return None

--# Usage
if job_response:
    status = check_ingestion_job_status(knowledge_base_id, data_source_id, ingestion_job_id)
    print(f"Current ingestion job status: {status}")

擬似コードを使用してデータをナレッジベースにプッシュする

疑似コードを使用して、使用可能なすべてのデータソースから取得したデータでナレッジベースを更新します。

例:

Function StartJob(knowledgeBaseId, dataSourceId)
    Try        job = BedrockAgentService.StartIngestionJob(knowledgeBaseId, dataSourceId)
        Return job
    Catch Error        LogError("Failed to start ingestion job for data source: " + dataSourceId)
        Return null
Function GetIngestionJobStatus(knowledgeBaseId, dataSourceId, ingestionJobId)
    Try        jobStatus = BedrockAgentService.GetIngestionJob(knowledgeBaseId, dataSourceId, ingestionJobId)
        Return jobStatus
    Catch Error        LogError("Failed to get status for job: " + ingestionJobId)
        Return null
Function RunIngestionJobs(knowledgeBaseId)
    dataSources = BedrockAgentService.ListDataSources(knowledgeBaseId)

    For Each dataSource in dataSources        job = StartJob(knowledgeBaseId, dataSource.Id)

        If job is not null Then            LogInfo("Job started successfully for data source: " + dataSource.Id)

            While job.Status is not (Completed or Failed or Stopped)
                Wait for short interval                job = GetIngestionJobStatus(knowledgeBaseId, dataSource.Id, job.Id)

                If job is null Then                    Break While loop
                        If job is not null Then                LogInfo("Job completed with status: " + job.Status)
            Else                LogError("Job monitoring failed for data source: " + dataSource.Id)
        Else            LogError("Failed to start job for data source: " + dataSource.Id)

Main    knowledgeBaseId = "<your-knowledge-base-id"
    RunIngestionJobs(knowledgeBaseId)

このコードでは、3 種類の主要関数を定義しています。

StartJob 関数は、StartIngestionJob API を使用して、指定したナレッジベースとデータソースの取り込みジョブを開始します。
GetIngestionJobStatus 関数は、指定した取り込みジョブの現在のステータスを取得します。
RunIngestionJobs 関数は、指定したナレッジベース内のデータソースの取り込みジョブを開始して監視します。

注: 操作が失敗した場合は、エラーメッセージで詳細を確認してください。

ベストプラクティスに従う

データを Amazon Bedrock に同期するときの問題を軽減するには、次のベストプラクティスを実施します。

API コールの問題を管理するために、プロセス全体にエラー処理を実装します。
ジョブのステータスを定期的に確認するために、ポーリングメカニズムを使用します。詳細については、「Lambda と AWS Batch を使用してジョブステータスをポーリングする」を参照してください。
トラブルシューティングや監査用に、取り込みプロセスの詳細なログを保持します。
セキュリティに関する AWS のベストプラクティスを実施します。
データ取り込みとストレージのコストをレビューします。

トピック: Machine Learning & AI Generative AI on AWS
タグ: Amazon Bedrock
言語: 日本語

データを Amazon Bedrock に自動的に同期させる方法を教えてください。

簡単な説明

解決策

IngestionJob API を使用する

擬似コードを使用してデータをナレッジベースにプッシュする

ベストプラクティスに従う

関連ビデオ

関連するコンテンツ