Amazon OpenSearch Ingestion で AWS Lambda をプロセッサとして使用し、埋め込みベクトルを生成する

本記事は、2025/1/21 に公開された Generate vector embeddings for your data using AWS Lambda as a processor for Amazon OpenSearch Ingestion を翻訳したものです。翻訳は Solutions Architect の山下一樹が担当しました。

2024 年 11 月 22 日、Amazon OpenSearch Ingestion が AWS Lambda プロセッサのサポートを開始しました。
この新機能の提供により、OpenSearch Ingestion パイプラインでログ、メトリクス、トレースデータを加工・変換する柔軟性が高まりました。
例えば、基盤モデル (FM) を使用してデータから埋め込みベクトルを生成したり、Amazon DynamoDB などの外部データソースを参照してデータをエンリッチできます。

Amazon OpenSearch Ingestion は、ログ、メトリクス、トレースデータをリアルタイムで Amazon OpenSearch Service ドメインと Amazon OpenSearch Serverless コレクションに配信する、完全マネージド型のサーバーレスデータパイプラインです。

プロセッサは、OpenSearch Ingestion パイプラインのコンポーネントで、目的の形式に変換した上で、指定した出力先にイベントを出力する前に、イベントをフィルタリング、変換、エンリッチできます。
パイプライン構成でプロセッサが定義されていない場合、イベントはソースコンポーネントで指定された形式で公開されます。
単一のパイプラインに複数のプロセッサを組み込むことができ、パイプライン構成で定義された順序で順次実行されます。

OpenSearch Ingestion では、データを変換する際に、ビルトインのネイティブプロセッサと共に Lambda 関数をプロセッサとして使用するオプションがあります。
イベントカウントまたはサイズに基づいて、イベントをまとめてバッチ的に Lambda を呼び出すことで、パフォーマンスとコストを最適化できます。
Lambda を使用すると、サーバーをプロビジョニングまたは管理する必要がなくなり、ワークロード量に応じてクラスターのサイズを変更するためのロジック、イベント統合の保守、ランタイムの管理が不要になります。

この投稿では、OpenSearch Ingestion の Lambda プロセッサを使用して、ソースデータの埋め込みを生成し、OpenSearch Serverless ベクトルコレクションに取り込む方法を示します。
このソリューションは、OpenSearch Ingestion パイプラインの柔軟性と Lambda プロセッサを組み合わせて、動的に埋め込みを生成します。
Lambda 関数は、Amazon Bedrock でホストされている Amazon Titan Text Embeddings Model を呼び出すため、効率的かつスケーラブルな埋め込み作成が可能です。
このアーキテクチャにより、レコメンデーションエンジン、パーソナライズされたチャットボット、不正検知システムなど、さまざまなユースケースの実装を簡単にします。

OpenSearch Ingestion、Lambda、OpenSearch Serverless を統合すると、文書埋め込み生成と検索のためのサーバーレスアプローチが提供されます。
この組み合わせにより、ワークロードの需要に合わせて自動的にスケーリングする、従量課金モデルが提供されます。
AWS がインフラストラクチャ、アップデート、メンテナンスを管理するため、運用が簡素化されます。
このサーバーレスアプローチにより、インフラストラクチャの管理ではなく、検索とアナリティクスソリューションの開発に集中できます。

Amazon OpenSearch Service は、ニューラル検索も提供しており、テキストをベクトル表現に変換し、テキストを取り込む際と検索時の両方でベクトル検索を容易にします。
テキストを取り込む際に、ニューラル検索はドキュメントテキストをベクトル表現に変換し、テキストとそのベクトル表現の両方をベクトルインデックスにインデックス化します。
バージョン 2.9 以上を実行するマネージドクラスターではニューラル検索を利用できます。

ソリューションの概要

このソリューションは、Amazon Simple Storage Service (Amazon S3) に保存されているデータセットから埋め込みベクトルを生成します。
OpenSearch Ingestion によって配信されたペイロードに対して、Amazon Titan モデルを適用するために Lambda 関数を使用します。

前提条件

Lambda 関数と Amazon Bedrock モデルを呼び出し、OpenSearch Serverless コレクションに書き込む適切な権限を持つロールが必要です。

コレクションにアクセスするには、コレクションへのアクセスを許可するアクセス許可ポリシーを持つ AWS Identity and Access Management (IAM) パイプラインロールを構成する必要があります。詳細については、Amazon OpenSearch Ingestion パイプラインにコレクションへのアクセスを許可するを参照してください。以下はコードの例です。

{
    "Version": "2012-10-17",
    "Statement": [ 
        {
            "Sid": "allowinvokeFunction",
            "Effect": "Allow",
            "Action": [ 
                "lambda:InvokeFunction"
                
            ],
            "Resource": "arn:aws:lambda:{{region}}:{{account-id}}:function:{{function-name}}"
            
        }
    ] 
}

このロールには、OpenSearch Ingestion がロールを引き受けることを許可する以下の信頼関係が必要です :

{
    "Version": "2012-10-17",
    "Statement": [ 
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "osis-pipelines.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ] 
}

OpenSearch Ingestion パイプラインの作成

ブループリントを使用してパイプラインを作成できます。この投稿では、AWS Lambda カスタムエンリッチメントブループリントを選択します。

データセットには、IMDB title basics dataset を使用します。このデータには、originalTitle、runtimeMinutes、ジャンルなどの映画情報が含まれています。

OpenSearch の取り込みパイプラインは、Lambda プロセッサを使用して original_title フィールドの埋め込みを作成し、その埋め込みを original_title_embeddings として他のデータと共に保存します。

次のパイプラインコードを参照してください :

version: "2"
 s3-log-pipeline:
  source:
    s3:
      acknowledgments: true 
      compression: "none"
      codec:
        csv:
      aws:
        # Provide the region to use for aws credentials 
        region: "us-west-2"
        # Provide the role to assume for requests to SQS and S3 
        sts_role_arn: "<<arn:aws:iam::123456789012:role/ Example-Role>>"
      scan:
        buckets:
          - bucket:
              name: "lambdaprocessorblog"
      
  processor:
     - aws_lambda:
        function_name: "generate_embeddings_bedrock"
        response_events_match: true 
        tags_on_failure: ["lambda_failure"] 
        batch:
          key_name: "documents"
          threshold:
            event_count: 4 
        aws:
          region: us-west-2 
          sts_role_arn: "<<arn:aws:iam::123456789012:role/Example-Role>>"
  sink:
    - opensearch:
        hosts:
          - 'https://myserverlesscollection.us-region.aoss.amazonaws.com'
        index: imdb-data-embeddings 
        aws:
          sts_role_arn: "<<arn:aws:iam::123456789012:role/Example-Role>>"
          region: us-west-2 
          serverless : true

OpenSearch Ingestion パイプラインの Lambda プロセッサをより詳しく見てみましょう。key_name パラメータに注目してください。key_name には任意の値を指定できますが、Lambda 関数では OpenSearch 取り込みからのペイロードを処理する際にこのキーを参照する必要があります。ペイロードのサイズはバッチ設定によって決まります。Lambda プロセッサでバッチ処理が有効になっている場合、OpenSearch 取り込みは複数のイベントをまとめて 1 つのペイロードにし、Lambda 関数を呼び出します。以下のいずれかの条件を満たすと、バッチが Lambda に送信されます。

- event_count – イベント数が指定された制限に達した時

- maximum_size – バッチの合計サイズが指定されたサイズ (例えば 5MB) に達した時。最大 6MB (AWS Lambda の呼び出し時のペイロードサイズの上限) まで設定可能

Lambda 関数

Lambda 関数は OpenSearch Ingestion からデータを受け取り、Amazon Bedrock を呼び出してベクトル埋め込み表現を生成し、ソースレコードにそれを追加します。documents は OpenSearch Ingestion から入ってくるイベントを参照するために使用され、パイプラインで宣言された key_name と一致します。Lambda 関数は、Amazon Bedrock からの埋め込みベクトルを元のレコードに追加します。この埋め込みベクトルが追加された新しいレコードは、OpenSearch Ingestion によって OpenSearch Serverless に出力されます。
次のコードを参照してください :

import json 
import boto3 
import os 

# Initialize Bedrock client 
bedrock = boto3.client('bedrock-runtime')

def generate_embedding(text):
    """Generate embedding for the given text using Bedrock."""
    response = bedrock.invoke_model(
        modelId="amazon.titan-embed-text-v1",
        contentType="application/json",
        accept="application/json",
        body=json.dumps({"inputText": text})
    )
    embedding = json.loads(response['body'].read())['embedding'] 
    return embedding 

def lambda_handler(event, context):
    # Assuming the input is a list of JSON documents 
    documents = event['documents'] 
    
    processed_documents = [] 
    
    for doc in documents:
        if 'originalTitle' in doc:
             # Generate embedding for the 'originalTitle' field 
            embedding = generate_embedding(doc['originalTitle'])
            
            # Add the embedding to the document 
            doc['originalTitle_embeddings'] = embedding 
        
        processed_documents.append(doc)
    
    # Return the processed documents 
    return  processed_documents

Lambda プロセッサを使用中に例外が発生した場合、バッチ内のすべてのドキュメントは失敗したイベントとみなされ、後続の処理フローがある場合はそちらに、なければ失敗と分かるように付与されたタグを付けて sink に転送されます。
このタグは、パイプラインの tags_on_failure パラメータで構成でき、エラーは CloudWatch ログにも送信されるため、さらなるアクションが可能です。

パイプラインの実行後、埋め込みが作成され、k-NN インデックスである imdb-data-embeddings 内のドキュメントに originalTitle_embeddings として格納されます。
次のスクリーンショットは、その例を示しています。

まとめ

この投稿では、OpenSearch Ingestion パイプラインの一部として Lambda を使用して、データの複雑な変換とエンリッチを可能にする方法を示しました。
この機能の詳細については、AWS Lambda を使用した OpenSearch Ingestion パイプラインの利用を参照してください。

著者について

Jagadish Kumar (Jag) は、Amazon OpenSearch Service に特化した AWS のシニアスペシャリストソリューションアーキテクトです。データアーキテクチャに情熱を持ち、AWS 上でアナリティクスソリューションを大規模に構築するお客様をサポートしています。

Sam Selvan は、Amazon OpenSearch Service の主任スペシャリストソリューションアーキテクトです。

Srikanth Govindarajan は、Amazon OpenSearch Service のソフトウェア開発エンジニアです。Srikanth は、検索、分析、セキュリティ、AI、機械学習ベースのユースケースのためのインフラストラクチャを設計し、スケーラブルなソリューションを構築することに情熱を持っています。

Cookie の設定を選択する

Amazon Web Services ブログ