Zenn
株式会社システムゼウスPublicationへの投稿
📑

AI が情報を検索して回答する仕組みを作る (2/2) | RAG で出典付きの回答を生成

めぐ
2025/03/11に公開
AWS
Bedrock
Pinecone
RAG
tech

この記事は前回の続きです。

はじめに

システムゼウスの池田です。
前回の記事では、Amazon Bedrock の Knowledge Bases + Agents を使って RAG（Retrieval-Augmented Generation、検索拡張生成）を構築しました。しかし、エージェントがナレッジベースから直接回答を生成する場合、出典情報が付かない という課題がありました。
本記事では、この課題を解決するために、アクショングループを活用し、出典付きの RAG を構築する方法 を紹介します。 具体的には、エージェントがナレッジベースを検索する際に AWS Lambda を介して検索結果に出典情報を付加し、エージェントへ返す形で実装します。

本記事の内容

  • Agents のアクショングループを使った RAG の実装方法
  • Knowledge Bases + Agents を活用した出典付き RAG の構築
  • Lambda 関数を用いた出典情報の付加
⭐こんな方におすすめ
  • Amazon Bedrock の Knowledge Bases や Agents を試してみたい人
  • 出典を表示できる RAG を構築したい人

💡アクショングループとは?

アクショングループとは、エージェントが関数や API を実行できるようにする仕組み です。設定すると、エージェントが状況に応じてアクションを実行できるようになります。1つのアクショングループには、複数のアクション（Lambda関数 や API）を含めることが可能[1]です。

例えば、アクショングループを使うことで以下のような処理ができます。

  • データベースや外部APIとの連携（リアルタイムの情報取得など）
  • 計算処理の実行（エージェントだけでは対応できない数値計算やデータ変換）
  • ナレッジベースの検索結果の加工（出典情報の付加など）

🔍 なぜアクショングループが必要なのか？

前回実装した方法では、エージェントは ナレッジベースの検索結果（文書のチャンク） しか受け取れず、出典情報を付けることができません。しかし、アクショングループを使って Lambda 関数を呼び出せば、検索結果とともに出典情報をエージェントに渡す処理を追加 できます。この違いを整理すると、エージェントに渡せる情報は以下のようになります。

方法 取得できるデータ 出典情報の有無
ナレッジベースのみ 関連する文書の一部 ❌（出典なし）
アクショングループ 関連する文書の一部 + 出典情報 ✅（出典あり）

この仕組みを活用することで、エージェントがより信頼性の高い回答を生成できます。検索結果の要約は、適切な回答が生成できない原因になる可能性があるため慎重に検討する必要があります。RAG では、検索結果は「関連する情報の断片の集合」であり、それらを保持することでエージェントが正確な回答を生成できます。

💡検索結果を要約せずにそのまま渡す理由

ベクトルデータ作成時に文書がチャンク（小さな単位）に分割されるため、LLM が要約した場合、意味を取り違えてしまう可能性があります。エージェントは、検索結果を基に最終的な回答を生成するため、検索結果はできるだけそのまま渡すのが望ましいです。また、特に技術文書や FAQ などでは、要約によって誤解を招くリスク もあるため、元の情報を保持することが重要になります。

この記事では、以下の手順で出典付きの検索結果を取得可能なアクショングループを作成し、エージェントが使えるようにします。

  1. Lambda 関数の実装
  2. エージェントの設定
  3. 出典付き RAG の動作検証

それでは、実装方法を詳しく見ていきましょう。

🪄Lambda 関数の実装

検索用の Lambda 関数を作成します。

1. インデックスへの接続情報のシークレットを作成

Pinecone のインデックスに接続するために必要な情報のシークレットを作成します。
キー/値のペアには以下を指定します。

キー 値
PineconeApiKey Pinecone のプロジェクトの API キー
PineconeHost 検索対象のインデックスの Host


Secret Managerで新しいシークレットを保存を選択し、図のように入力


シークレットの名前と説明を入力し保存


作成されたシークレットの画面。 4. Lambda 関数の設定でこの ARN を使用する

2. Lambda の実行ロールの作成

コンテナイメージで Python Lambda 関数をデプロイするため、実行ロールを作成します。


「 IAM 」で「ロールを作成」を選択し、上記の内容を入力

CloudWatch等を使えるようにするために必要な「 AWSLambdaBasicExecutionRole 」を選択します。


Lambda で使用されるポリシー「 AWSLambdaBasicExecutionRole 」を選択


確認し、ロールの作成を完了

続いて、作成したロールに関数内で使用するサービスについてのアクセス許可を追加します。Bedrock の基盤モデル呼び出しと Secrets Mangaer の値の取得を追加します。

「許可ポリシー」の「インラインポリシーを作成」を選択

以下の表・図のように設定します。

サービス アクション リソース リソースARN
Bedrock InvokeModel foundation-model arn:aws:bedrock:{リージョン}::foundation-model/amazon.titan-embed-text-v1
Secrets Manager GetSecretValue Secret 1.で作成したシークレットの ARN


表のとおりに入力した画面


ポリシー名を付けて作成を完了する

3. Lambda 関数の作成

「 Python の AWS ベースイメージを使用する」手順に従って Lambda 関数をコンテナイメージでデプロイしていきます。

  • ベースイメージからイメージを作成

lambda_function.py、requirements.txt、Dockerfile は以下の内容にします。

lambda_function.py
lambda_function.py
import json
import os
import boto3
import sys
from pinecone.grpc import PineconeGRPC
from typing import List, Dict
from pinecone.grpc import GRPCIndex

session = boto3.Session(region_name="ap-northeast-1")
client_bedrock = session.client(service_name="bedrock-runtime")
client_lambda = session.client(service_name="lambda")
secrets_manager = session.client("secretsmanager")
pinecone_secret_arn = os.environ["PineconeSecretArn"]
model_id = "amazon.titan-embed-text-v1"

"""
・処理の流れ
eventから検索クエリを取得
検索クエリから密ベクトルを生成（Titan利用）
search_knowledgebase関数を実行
    Pineconeに対して密ベクトルで検索を実行
    データサイズを適切に調整
データをAgentに返却する形式に整形
Agentに取得したテキストデータを返却
"""


def handler(event, context):
    try:
        # Secret Managerからシークレットを取得
        Pinecone_API_KEY = get_secret(
            pinecone_secret_arn,
            secrets_manager,
        )["PineconeApiKey"]
        Pinecone_HOST = get_secret(
            pinecone_secret_arn,
            secrets_manager,
        )["PineconeHost"]

        print(f"Pinecone_API_KEY: {Pinecone_API_KEY}")
        print(f"Pinecone_HOST: {Pinecone_HOST}")

        # eventから各種パラメータの取り出し
        actionGroup = event["actionGroup"]
        function = event["function"]
        parameters = event.get("parameters", [])

        # parametersからクエリを取り出す（最初の要素のvalueキーの内容がクエリ）
        if parameters and "value" in parameters[0]:
            query = parameters[0]["value"]
            print(f"Value(query): {query}")
        else:
            query = None
            print("No value found in parameters(query)")

        if not query:
            raise ValueError("queryが指定されていません")

        # 検索クエリから密ベクトルを生成（Titanを利用）
        native_request = {"inputText": query}
        request = json.dumps(native_request)

        try:
            response = client_bedrock.invoke_model(
                modelId=model_id, body=request
            )
        except Exception as e:
            raise e

        response_body = response["body"].read().decode("utf-8")
        response_data = json.loads(response_body)

        # 密ベクトルを取得
        dense_vector: List[float] = response_data["embedding"]
        # print({f"dense_vector: {len(dense_vector)}"})

        # Pineconeに対して検索を実行し、整形した結果を取得する
        pc = PineconeGRPC(api_key=Pinecone_API_KEY)

        index = pc.Index(host=Pinecone_HOST)

        try:
            result_dict = search_knowledgebase(
                index, dense_vector
            )
        except Exception as e:
            raise e

        # Agentに返却する形式に整形
        combined_results_json = json.dumps(result_dict, ensure_ascii=False)

        body = {"TEXT": {"body": combined_results_json}}

        action_response = {
            "actionGroup": actionGroup,
            "function": function,
            "functionResponse": {"responseBody": body},
        }

        func_response = {
            "response": action_response,
            "messageVersion": event["messageVersion"],
        }

        print(f"func_response: {func_response}")
        print(f"size of func_response: {sys.getsizeof(func_response)}")

        return func_response

    except Exception as e:
        return {
            "statusCode": 500,
            "body": json.dumps({"message": str(e)}),
        }


def search_knowledgebase(
    index: GRPCIndex,
    dense_vector: List[float],
) -> List[Dict[str, str]]:
    """
    Pineconeに対して密ベクトルで検索を実行し、結果を取得する

    Args:
        index (GRPCIndex): PineconeのIndexにアクセスするためのインスタンス
        dense_vector (List[float]): 密ベクトル

    Returns:
        List[Dict[str, str]]: 本文とソースの辞書のリスト

    example:
    ```
    result_list = [
        {"source": "source1", "text": "text1"},
        {"source": "source2", "text": "text2"},
        ...
        ]
    ```
    """

    try:
        result_dict: List[Dict[str, str]] = []

        # top_kを5から50まで5刻みで検索
        for top_k in range(5, 50, 5):
            # 検索を実行
            knowledgebase_search_results = index.query(
                top_k=top_k,
                vector=dense_vector,
                include_metadata=True,
            )

            # QueryResponse を辞書に変換
            search_results_dict = knowledgebase_search_results.to_dict()

            # `matches` にアクセスし、結果を取得
            matches = search_results_dict.get("matches", [])
            if not matches:
                print("Warning: No matches found in Pinecone query.")
                continue

            for match in matches:
                metadata = match.get("metadata", {})
                metadata_str = json.dumps(
                    metadata, indent=2, ensure_ascii=False
                )
                print(f"Metadata: {metadata_str}")

                # `metadata` キーの値が JSON 文字列の場合はデコードする
                source = "unknown"
                if "metadata" in metadata:
                    try:
                        decoded_metadata = json.loads(metadata["metadata"])
                        source = decoded_metadata.get("source", "unknown")
                    except json.JSONDecodeError:
                        print("Warning: Failed to decode 'metadata' as JSON.")

                # `text` の取得
                text = metadata.get("text", "No text found")

                print(f"Extracted source: {source}")
                print(f"Extracted text: {text}")

                result_dict.append({
                    "source": source,
                    "text": text,
                })

                print(f"result_dict: {result_dict}")

            # result_dictの中に5種類以上のparentTextがあるならループ終了
            if len(result_dict) >= 5:
                break

        # レスポンスサイズをチェックし、25KB(エージェントへのレスポンスサイズ上限)を超えないように調整
        result_dict = check_response_size(result_dict)

        return result_dict

    except Exception as e:
        message = error_message(e, "knowledgebase_search")
        print(message)
        raise Exception(message)


def get_secret(secret_arn, secrets_manager):
    try:
        response = secrets_manager.get_secret_value(SecretId=secret_arn)
        if "SecretString" in response:
            return json.loads(response["SecretString"])
        else:
            return json.loads(response["SecretBinary"])
    except Exception as e:
        print(f"Error retrieving secret: {str(e)}")
        raise e


def check_response_size(
    response_data: List[Dict[str, str]],
    max_size_kb: int = 25
) -> List[Dict[str, str]]:
    """
    レスポンスデータのサイズをチェックし、指定されたサイズを超える場合はデータ量を減らす

    Args:
        response_data (List[Dict[str, str]]): チェックするレスポンスデータ
        max_size_kb (int): 最大サイズ（KB）

    Returns:
        List[Dict[str, str]]: サイズを超えないように調整されたレスポンスデータ
    """
    max_size_bytes = max_size_kb * 1024
    current_size_bytes = len(json.dumps(response_data).encode('utf-8'))

    while current_size_bytes > max_size_bytes and len(response_data) > 1:
        response_data.pop()  # 末尾の要素を削除
        current_size_bytes = len(json.dumps(response_data).encode('utf-8'))
    print("check_response_size: ", current_size_bytes)
    return response_data


def error_message(e: BaseException, message: str) -> str:
    """
    エラー出力メッセージを作成

    Args:
        e (BaseException): エラー
        message (str): メッセージ

    Returns:
        str: エラーメッセージ
    """
    errorType = type(e).__name__

    error_message = f"{errorType} occurred: {str(e)}\n {message}"

    return error_message
requirements.txt
requirements.txt
pinecone[grpc]
boto3
Dockerfile
Dockerfile
FROM public.ecr.aws/lambda/python:3.12

# Copy requirements.txt
COPY requirements.txt ${LAMBDA_TASK_ROOT}

# Install the specified packages
RUN pip install -r requirements.txt

# Copy function code
COPY lambda_function.py ${LAMBDA_TASK_ROOT}

# Set the CMD to your handler (could also be done as a parameter override outside of the Dockerfile)
CMD [ "lambda_function.handler" ]
  • Lambda 関数の内容
    コードの処理フロー
    1. 検索クエリを取得（エージェントのパラメータから検索クエリを受け取る）
    2. Titanを使って密ベクトルを生成（検索クエリをベクトル化）
    3. Pineconeで検索（密ベクトルで検索を実行し、最も関連性の高い文書5件を取得）
    4. 検索結果に出典情報を追加（メタデータの source を抽出し、エージェントに渡す）
    5. エージェントへ返却（検索結果 + 出典情報をJSONで返す）

    エージェントが検索クエリを指定して実行すると、検索結果と検索結果のファイルの場所を以下の形式で返す関数となっています。

    エージェントへ返す値
    {
    'response': {
        'actionGroup': 'action_group_search_new_manual',
        'function': 'search_new_manual',
        'functionResponse': {
            'responseBody': {
                'TEXT': {
                    'body': '[
                                {"source": "検索結果1のファイルの場所", "text": "検索結果1"}, 
                                {"source": "検索結果2のファイルの場所", "text": "検索結果2"}, 
                                {"source": "検索結果3のファイルの場所", "text": "検索結果3"}, 
                                {"source": "検索結果4のファイルの場所", "text": "検索結果4"}, 
                                {"source": "検索結果5のファイルの場所", "text": "検索結果5"}
                            ]'
                }
            }
        }
    },
    'messageVersion': '1.0'
}
    🔎Lambda 関数のカスタマイズ例

    検索方法や結果の加工方法を変えることもできます。

    • ハイブリッド検索
    • 絞り込み検索
    • 独自で定義したメタデータの返却

    以下を追加で行うと、上記の拡張が可能です。

    • ハイブリッド検索

      • ベクトルデータベースにチャンクを疎ベクトル化したものを格納する
      • 検索クエリを疎ベクトル化して検索で使用する

    • 絞り込み検索・独自で定義したメタデータの返却

  • イメージのデプロイ

Lambda 関数を作成の際には、ロールに2.で作成したロールの ARN を指定します。

Lambda 関数作成コマンドの例
aws lambda create-function \
  --function-name hello-world \
  --package-type Image \
  --code ImageUri=111122223333.dkr.ecr.us-east-1.amazonaws.com/hello-world:latest \
  --role <2.で作成したロールのARN>

4. Lambda 関数の設定を追加

コンソールを開き、作成した Lambda の設定を以下のように追加します。

  • 一般設定
    タイムアウトを1分に設定します。

  • 環境変数
    以下の値に設定します。

    キー 値
    PineconeSecretArn インデックスへの接続情報のシークレットの ARN（1.で作成したもの）

  • アクセス権限のリソースベースのポリシーステートメント
    エージェントからの呼び出し許可を追加します。
    「アクセス許可」の「リソースベースのポリシーステートメント」で「アクセス権限を追加」を選択し、以下の内容を入力します。
    ソース ARN には、エージェント ARN を指定します（エージェントの概要から確認できます）。

    サービス プリンシパル ソース ARN アクション
    Other bedrock.amazonaws.com エージェントの ARN lambda:InvokeFunction


    表のように入力した画面

🔧エージェントの設定

エージェントが検索したい時に Lambda を呼び出せるように設定していきます。
エージェントの概要の画面で「エージェントビルダーで編集」を選択します。

1. Lambdaをアクションとして登録

  1. 「アクショングループ」の「追加」を選択します。

  2. 「アクショングループの詳細」のアクショングループ名を入力します。

  3. 「アクショングループの呼び出し」の「既存の Lambda 関数を選択」を選択し、先ほど作成した Lambda 関数を指定します。

  4. アクショングループ関数の設定では、名前を入力し、パラメータを以下のように設定します。

    名前 説明 タイプ 必須
    query 検索クエリ String True


    表のように入力した画面

この設定で、エージェントはアクションとして登録した Lambda 関数を状況に応じて必要なパラメータを渡し、使えるようになります。

2. ナレッジベースを無効化

「ナレッジベース」の「状態」を「 DISABLED 」に変更します。


現在あるナレッジベースを無効化

🔎ナレッジベースを無効化する理由
  • エージェントが直接ナレッジベースを使うと出典情報を得ることができない
  • ナレッジベースを無効化することで、エージェントが直接ナレッジベースを使わないようにする
  • 無効化しなくてもエージェントが判断してアクショングループを実行してくれることが多いが念のため

3. エージェントへの指示を更新

検索する必要があると判断した場合はアクションを実行するように指示します。


エージェントへの指示

全ての設定が完了したら、「保存して終了」をクリックし、保存します。

🤖出典付き RAG の動作検証

「テスト」の「準備」をクリックしてエージェントの設定を更新し、テストエージェントに質問します。


テストエージェントの実行結果

出典付きで回答できていることが分かります。

🛠 エラーと解決策

✅ Lambda関数がエージェントから呼び出せない

  • 「リソースベースのポリシーステートメント」の設定を再確認
  • アクショングループの設定を再確認
  • CloudWatch Logs を確認し、Lambda 関数のエラーやリクエストパラメータをチェック

✅ Lambda関数でエラーが発生している

  • エージェントへのレスポンスが正しい形式になっているか確認

✅ Pinecone の検索が動作しない

  • Secret Manager の値を確認し、環境変数 PineconeSecretArn が正しいか確認
  • 検索クエリから密ベクトルの生成がうまくできているか確認

🔎トレースの確認

テストエージェントのトレースを確認し、エージェントの実行の流れを追ってみます。


テストエージェントのトレース

トレースステップ1に着目すると、エージェントは以下を行っていることが分かります。

  1. ユーザーからの質問から、検索する必要があることを判断
  2. 適切な検索クエリを生成
  3. 検索クエリを引数としてアクショングループ「action_group_search_new_manual」を実行

トレースステップ2に着目すると、エージェントは以下を行っていることが分かります。

  1. アクショングループの実行結果を受け取る[2]
  2. 実行結果を基に回答を生成（検索結果のファイル名を出典として含める）

一連のステップによって、ナレッジベースをエージェントに直接繋がずに出典付きの回答を生成していることが分かります。

📝まとめ

本記事では、Amazon Bedrock の Knowledge Bases + Agents を活用し、出典付きで回答を生成する方法について解説しました。従来のナレッジベースだけでは出典情報が付かないという課題に対し、アクショングループを利用して Lambda 関数を呼び出し、検索結果に出典を付与する手法を実装しました。

🎯本記事のポイント
  • ナレッジベースのみでは出典が付かない課題
    → 検索結果の信頼性を高めるために出典情報を付与する必要がある
  • アクショングループを活用した Lambda 関数の実装
    → エージェントが直接検索するのではなく、Lambda 関数を経由して出典付きの検索結果を取得
  • エージェントの設定変更（ナレッジベースの無効化 & アクションの追加）
    → エージェントが適切に Lambda を呼び出せるように設定
✅実装のメリット
  • 出典付きの回答により、ユーザーが情報の信頼性を判断しやすくなる
  • ナレッジベースの検索結果を加工できるため、柔軟な拡張が可能
  • アクショングループ・ Lambda 関数を活用することで、検索処理のロジックを独自に制御できる

本手法を応用すれば、検索結果のフィルタリングや外部APIと組み合わせた高度な回答生成も実現可能です。「出典付きの回答」を求めるユースケースがある場合、ぜひ今回の方法を参考にしてみてください。

免責事項

作者または著作権者は、契約行為、不法行為、またはそれ以外であろうと、ソフトウェアに起因または関連し、あるいはソフトウェアの使用またはその他の扱いによって生じる一切の請求、損害、その他の義務について何らの責任も負わないものとします。

脚注

  1. 複数のAPIオペレーションを含めることができますが、記述できるLambda関数は1つだけです（注記を参照）。 ↩︎

  2. 検索結果(text)と検索結果のファイルの場所(source)のペアがトレースステップ2に含まれていることから分かります ↩︎

株式会社システムゼウス
株式会社システムゼウスPublication

アプリケーションの受託開発が主事業です。クラウドやIoT関連の開発を得意としています。 「やってもいないことをデキると言わない」をモットーに、ここではエンジニアが現場で得た知見や、実際に実験した内容などを発信します。

Discussion