オンデマンドおよびバッチパイプラインによる動的なデータ抽出

多くの企業には、未活用ビジネスインテリジェンスを含む大量の紙文書または電子文書が存在します。[生成 AI](https://aws.amazon.com/generative-ai/) の進展に伴い、さまざまな大規模言語モデル（LLM）を用いて、これらの文書から関連データを正確に抽出することが可能になりました。本記事では、ドキュメント処理時間とコストの柔軟性を確保するため、オンデマンド推論とバッチ推論の両方のオプションを備えたインテリジェントなドキュメント処理パイプラインについて解説します。[Amazon Bedrock](https://aws.amazon.com/bedrock/) 上で動作するこのパイプラインでは、時間制約のあるリクエストにはオンデマンド推論オプションを、コスト最適化が最優先されるケースにはバッチ推論オプションを使用できます。また、ドキュメントレベルで大規模言語モデルとプロンプトを動的に指定する方法についても説明し、同じパイプラインを用いて複数の種類の文書からデータを抽出できるようにします。 ## ソリューション概要 お客様の中にも、スキャンされた PDF 形式（編集可能なテキストを含まず画像のみを含む PDF、例えばこの場合はスキャンされた土地賃貸契約書が PDF として保存されているケース）で数百万件の文書をバックログとして抱え、さらに新しい文書が毎日積み上がっているという方がいらっしゃるかもしれません。そのような場合でも、本ソリューションを使用すれば、これらの文書から効果的にデータを抽出することが可能です。以下の図に示す通り、このソリューションはオンデマンド型とバッチ型の 2 つの推論パイプラインを構築し、動的に呼び出す仕組みを提供します。Amazon Bedrock [プロンプト管理](https://aws.amazon.com/bedrock/prompt-management/) で効果的に設計・管理されたプロンプトを活用することで、形式や規約がバラバラであることが多いスキャン PDF やテキストファイルからデータを抽出し、標準化することができます。 ![image](https://d2908q01vomqb2.cloudfront.net/f1f836cb4ea6efb2a0b1b99f41ad8b103eff4b59/2026/06/02/ML-18264-1.png) 左側のパイプラインはオンデマンド型で、文書を 1 つずつ抽出し、数秒以内に結果を返します。このため、時間制約のあるリクエストに適しています。 右側のパイプラインは、単一の Amazon Bedrock [バッチ推論](https://docs.aws.amazon.com/bedrock/latest/userguide/batch-inference-create.html) ジョブで複数のドキュメントリクエストを処理するバッチ推論パイプラインであり、モデルの呼び出しは非同期で処理されます。ユーザーは両方のパイプラインにおいて、リクエスト内でプロンプト ID とバージョンを指定でき、対応するプロンプトテキストは Amazon Bedrock プロンプト管理から取得されます。 以下のセクションでは、両方のパイプラインの詳細な説明を提供します。 ## 1. オンデマンド推論パイプライン オンデマンド推論パイプラインには、[AWS SQS](https://aws.amazon.com/sqs/) の First-In, First-Out (FIFO) キューが作成されます。ドキュメント ID、LLM モデル ID、プロンプト ID/バージョン、システムプロンプト ID/バージョンを含むキューメッセージが到着すると、AWS Lambda 関数がトリガーされます。この関数は、指定された Amazon S3 バケットから PDF ドキュメントを取得し、PDF ページを PNG 画像に変換し、Amazon Bedrock プロンプト管理から関連するプロンプトを取得し、LLM を呼び出すためのメッセージを組み立て、その結果を [Amazon DynamoDB](https://aws.amazon.com/dynamodb/) テーブルに保存します。 ## 1.1. AWS SQS FIFO キュー 単一のドキュメントが到着した際に Amazon Bedrock の推論をトリガーするために、AWS SQS FIFO キューが使用されます。FIFO キューを使用する主な理由は以下の通りです： - リライアブルメッセージ配信 – 各メッセージが正確に一度だけ配信されることを保証します。 - First-In, First-Out (FIFO) プロセッシング – 厳格な順序を維持し、処理の予測可能性を高めます。 - メッセージグループ化 – Message Group ID 属性により、各グループ内でメッセージが順序通りに処理されることが保証されます。各プロデューサーは、関連するメッセージの順序を維持するために一意の Message Group ID を使用できます。 ## How is a queue message created? キューメッセージは、AWS CLI または AWS SDK API を用いて外部から作成できます。以下に AWS CLI コマンドの例を示します: aws sqs send-message --queue-url https://sqs.us-east-1.amazonaws.com/1111111111/ondemand-data-pipeline-queue.fifo --message-group-id "1" --message-body "msg 1" --message-attributes file://message_txt.txt この例におけるファイル message_txt.txt は、アプリケーションに必要なメッセージ属性を含む JSON ファイルです。詳細は以下の「パイプラインのテスト」セクションをご覧ください。 Lambda 関数は、Amazon Bedrock が抽出データを返した後、キューメッセージを削除します。 ## 1.2. Lambda function – queue message processing and inferencing ## 1.2.1 Retrieving the documents, converting to images, and splitting large files Lambda 関数は、キューメッセージ内の s3_location 属性を使用してドキュメントをダウンロードします。ドキュメントがスキャンされた PDF の場合、マルチモーダルモデルが理解できるように画像に変換されます。 執筆時点において、Claude 4 Sonnet モデルはマルチモーダル呼び出しあたり最大 20 枚までの画像しか許可していません。したがって、ドキュメントに 20 ページを超える画像が含まれる場合は、20 ページごとのチャンクに分割する必要があります。doc_id（文書識別子）、chunk_count（チャンク数）、chunk_id（各チャンクの識別子）は、抽出結果およびモデルのパフォーマンス指標とともに Amazon DynamoDB テーブルに格納されます。 - doc_id: 文書の識別子 - chunk_count: その文書に対するチャンクの総数 - chunk_id: 文書の各チャンクの識別子 ## 1.2.2. Amazon Bedrock Prompt Management からのプロンプト取得 土地賃貸契約書はフォーマットが様々です。番号付きリストで土地区画の属性を示すものもあれば、表形式のもの、さらには土地図面として記述されたものもあります。そのため、各ドキュメントのフォーマットに合わせた異なるプロンプトを使用することで、抽出精度を向上させることができます。 LLM 呼び出しで使用されるプロンプトは Amazon Bedrock Prompt Management に格納されています。各プロンプトには一意の ID が割り当てられ、バージョン管理されています。SQS メッセージでは関連するプロンプト ID とバージョンを指定する必要があり、Lambda 実行時にこれらの情報を用いてプロンプト本文が取得されます。 注記：リージョンあたり最大 50 プロンプト、プロンプトあたり最大 10 バージョンというサービス制限があります。 ## 1.2.3 LLM 呼び出し用のメッセージ作成とレスポンス処理 Lambda 関数は以下の手順を続行します： - LLM のメッセージは、プロンプト本文と画像を連結して作成します。 - Converse API を使用して Amazon Bedrock へリクエストを送信します。 LLM は抽出データを JSON 文字列として返します。この結果は、以下のパイプラインテストセクションで示すように、DynamoDB テーブルで確認できます。 ## 1.2.4 結果の保存 最後に、Lambda 関数は以下の処理によってプロセスを完了します： - JSON を解析し、土地区画の属性を DynamoDB テーブルに格納します。 - ドキュメントが正常に処理され、結果が保存された場合、SQS メッセージはキューから削除されます。 ## 2. バッチ推論パイプライン バッチ推論パイプラインには、高いスループットを実現するために標準的な AWS SQS キューが使用されます。メッセージの作成方法はオンデマンドパイプラインと同様ですが、message-group-id 属性は不要です。 バッチ推論パイプラインの主要なコンポーネントは以下の通りです： - Amazon EventBridge Scheduler（スケジューラ）。 - スキャンされた PDF の前処理、JSONL ファイルの作成、およびバッチ推論ジョブの送信を行う Batch Inference AWS Lambda 関数。 - Amazon EventBridge ルール。 - 後処理を行う Post-processing AWS Lambda 関数。 以下のセクションでバッチ推論パイプラインの詳細を説明します。 ## 2.1. Amazon EventBridge スケジューラ Amazon EventBridge Scheduler は、スケジュールに基づいてバッチ推論用 Lambda 関数を起動します。 ## 2.2. バッチ推論 Lambda 関数 この関数は、処理を続行する前にキュー内に十分なメッセージがあるかどうかをまず確認します。執筆時点では、Amazon Bedrock のバッチ推論ジョブに対して、レコードの最小数が 100 件と設定されています。 ## 2.2.1 キューからのメッセージ受信 Lambda 関数はキュー内のメッセージをループ処理し、ドキュメント ID、LLM モデル ID、プロンプト ID/バージョン、およびシステムプロンプト ID/バージョンを抽出します。 ## 2.2.2 ドキュメントの重複排除取得、画像への変換、大規模ファイルの分割 次に Lambda 関数はドキュメントを取得し、スキャンされた PDF の場合は画像に変換し、必要に応じて大規模ファイルを分割します。これはオンデマンドパイプラインと同様の処理です。標準的な SQS キューは正確に一度だけのメッセージ配信を保証しないため、この関数では重複するメッセージも無視されるようにしています。 ## 2.2.3 バッチ推論ジョブ内での異なるプロンプトの許可 オンデマンドパイプラインと同様に、より効果的なデータ抽出のためには、異なるドキュメント形式に対して異なるユーザープロンプトが必要です。 各ドキュメントに対する意図するプロンプト ID とバージョンは SQS メッセージ内で指定されています。Lambda 実行中、関数は Amazon Bedrock のプロンプト管理からプロンプト本文を取得します。 ## 2.2.4 バッチ推論ジョブ用の JSONL アーティファクト作成 その後、Lambda 関数は以下のタスクを処理します： - バッチ推論データ用 S3 バケットに metadata.json を作成し、SQS メッセージ ID、ドキュメント ID、プロンプト ID/バージョン、システムプロンプト ID/バージョン、およびその他のプロジェクト関連属性を含むメッセージ属性を保存します。このファイルは後で Post-Processing Lambda によって DynamoDB テーブルの埋め込みに使用されます。 - Amazon Bedrock のバッチ推論ジョブに必要な JSONL ファイルを作成するためにドキュメントを処理します。このプロセスは効率化のために Python の multiprocessing モジュールを使用して並列化され、JSONL ファイルはバッチ推論データ用 S3 バケットにアップロードされます。 - ドキュメントの準備と S3 バケットへのアップロードが完了した後、SQS メッセージを削除します。これにはキューに対して大きな可視性タイムアウトを設定する必要があります。 ## 2.2.5 メッセージの作成とバッチ推論ジョブの送信 最後に、バッチ推論 Lambda 関数は、前工程で生成された JSONL アーティファクトを使用して Amazon Bedrock のバッチ推論ジョブを作成します。各バッチジョブは 1 つのモデルのみを使用してドキュメントを処理できるため、同じバッチジョブ内の SQS メッセージはすべて同一のモデル ID を指定する必要があります。入力メッセージに複数のモデル ID が指定されている場合、Lambda 関数はポーリングメカニズムを使用し、最も頻繁に指定されたモデル ID を選択して使用します。 ## 2.3. Amazon Bedrock バッチ推論ジョブ Amazon Bedrock がバッチ推論ジョブを受け取ると、それをキューに配置します。ジョブが開始されると、以下の手順に従って処理が進みます。 ## 2.3.1 バッチ推論ジョブ用の JSONL アーティファクトの取得 Amazon Bedrock は、ジョブ作成時に指定された JSONL アーティファクトを取得します。 ## 2.3.2 バッチ推論出力の保存 完了後、Amazon Bedrock は、ジョブ作成時にも指定されたバッチ推論データ用 S3 バケットへ出力を保存します。 ## 2.3.3 Amazon EventBridge への通知 ジョブ完了後、Amazon Bedrock はイベントブランチルールによって捕捉される、Amazon EventBridge 宛てのジョブステータス変更イベントを送信します。 ## 2.4. Amazon EventBridge ルールが推論後の Lambda 関数をトリガー EventBridge ルールは、追加のモデル出力処理を処理するための推論後処理用 Lambda 関数をトリガーします。 ## 2.5. 推論後処理 Lambda 関数 ## 2.5.1 出力 JSONL の取得 Lambda 関数は、バッチ推論データ用 S3 バケットから推論出力の JSONL をフェッチします。 ## 2.5.2 推論出力の保存 この関数は JSONL ファイルを解析し、抽出された土地区画属性を DynamoDB テーブルに保存します。 ## 前提条件 この例を自分で試したい場合は、以下の前提条件を満たしていることを確認してください: - AWS Management Console にアクセスできる AWS アカウント - CloudFormation スタックの作成と管理に必要な適切な IAM 権限。通常、以下が含まれます: cloudformation:CreateStack - cloudformation:DescribeStacks - cloudformation:UpdateStack - cloudformation:DeleteStack ## CloudFormation スタックのデプロイ オンデマンドパイプラインをデプロイします: Launch Stack リンクを選択すると、[AWS CloudFormation](https://aws.amazon.com/cloudformation/) に遷移し、CloudFormation スタックが起動されます: - Create stack ページで Next を選択します。 - Specify stack details ページで Next を選択します。 - Configure stack options ページで Next を選択します。 - Review and create ページで、AWS CloudFormation が IAM リソースを作成する可能性があることを承認するチェックボックスを選択します。 - Submit を選択します。 送信後、Stack info、Events、Resource など、スタックに関するいくつかの詳細を確認できます。以下にイベントのスクリーンショットを参考として示します: ![image](https://d2908q01vomqb2.cloudfront.net/f1f836cb4ea6efb2a0b1b99f41ad8b103eff4b59/2026/06/02/ML-18264-4.png) 同様の手順でバッチパイプラインもデプロイできます。 ## パイプラインのテスト 以下の手順は、オンデマンドパイプラインをテストするためのガイドです。少なくとも 100 ドキュメントをお持ちであれば、バッチパイプラインも同様の手順でテスト可能です。 - ローカル環境にデータをダウンロードします。テキサス州土地記録および郡記録ウェブサイトから購入した、ウィンクラー郡、アンドリュース郡、サットン郡の 3 つの土地文書があります。 - ダウンロードした PDF ファイルを、CloudFormation スタックで作成されたオンデマンドデータパイプラインバケット `on-demand-data-pipeline-bucket-${account_id}` にアップロードします。 - 以下の例を元に、CloudFormation スタックから作成されたプロンプト ID、システムプロンプト ID、および S3 バケットを置き換えて、テキストファイル `message_txt.json` を作成してください。 ```json { "application": { "DataType": "String", "StringValue": "bedrock-example" }, "id": { "DataType": "String", "StringValue": "Winkler_2024-06-05_N_C42758_V_OPR" }, "model_id": { "DataType": "String", "StringValue": "anthropic.claude-sonnet-4-20250514-v1:0" }, "prompt_id": { "DataType": "String", "StringValue": "6CT88W3MWT" }, "prompt_version": { "DataType": "String", "StringValue": "1" }, "s3_location": { "DataType": "String", "StringValue": "s3://ondemand-data-pipeline-bucket-111111111/Winkler_2024-06-05_N_C42758_V_OPR.pdf" }, "system_prompt_id": { "DataType": "String", "StringValue": "R2NFLXFXOJ" }, "system_prompt_version": { "DataType": "String", "StringValue": "1" } } 翻訳全文 - 上記の AWS CLI の例を用いて、キュー名を置き換えたシェルスクリプト send2queue.sh を作成し、実行してください。すると、SQS キュー ondemand-data-pipeline-queue.fifo にメッセージが送信されたことを示す通知が表示されます。 - このキューメッセージは、Lambda 関数 ondemand-data-pipeline-queue-processor をトリガーします。 - Amazon CloudWatch の Lambda ログを確認してください。ロググループ名は /aws/lambda/ondemand-data-pipeline-queue-processor です。 - DynamoDB テーブル ondemand-data-pipeline-table 内の Amazon Bedrock 推論出力を確認してください。Winkler カウンティの例における model_response カラムの JSON 結果は、以下のようになります: [ { "tract": 1, "state": "Texas", "county": "Winkler", "abstract": "A-1239", "survey": "PSL Survey", "section": "8", "range_block": "B2", "quarter": "N/2 of N/2" }, { "tract": 2, "state": "Texas", "co

オンデマンドおよびバッチパイプラインによる動的なデータ抽出

背景や根拠まで確認しますか？

調べる

選ぶ

サイト