Guardrails for Amazon Bedrock で遊んでみる #AWS

はじめに

2024/4/24 に降り注いできた Amazon Bedrock の大量アップデートで、re:Invent 2023 で発表されていた Guardrails for Amazon Bedrock がバージニア北部とオレゴンリージョンで一般利用可能になりました。

アプリケーションにとって望ましくない、または有害なカテゴリに分類されるプロンプトや応答を回避するために複数のポリシーを組みあわせてガードレールを作成することができます。

ガードレールはモデルとアプリケーションの間に介入するため、既存の構成に組み込みやすく、また Bedrock で利用可能な多くのモデルにも対応しています。

ガードレールの作成

以下の4つのポリシーを任意に組み合わせて (有効化/無効化して) ガードレールを構成できます。

コンテンツフィルター
拒否されたトピック
ワードフィルター
機密情報フィルター

コンテンツフィルター

特定のカテゴリーに対してフィルターの強度を調整して、有害なコンテンツを含む入力プロンプトまたはモデル応答をフィルターできます。近しい機能は Azure OpenAI Service にもあったような気がしますね。

拒否されたトピック

アプリケーションのコンテキストにおいて望ましくない話題を定義し、応答を拒否できます。ドキュメントでは銀行アシスタントに、投資アドバイスにやマネーロンダリングに関連する会話を避けるといったユースケースが記載されていました。

トピックを自然言語で定義し、トピックに関連するサンプルフレーズをいくつか指定することで定義できます。

ワードフィルター

プロンプトや応答に含まれる可能性のある望ましくない単語やフレーズ、冒涜的な表現などをブロックできます。冒涜表現は事前定義され、多言語に対応したリストを使用できるほか、手動またはファイルアップロードで単語や3単語までのフレーズを一括登録できます。

機密情報フィルター

事前定義された個人情報 (PII) をブロックまたはマスクできます。事前定義されていない情報に関しては正規表現を使用してブロックまたはマスクすることも可能です。

ガードレールのテストとデプロイ

テスト

作成したガードレールはコンソール上でテストを実行できます。素のモデルの応答と Guardrail による最終応答を確認でき、どのポリシーで拒否されたかまで確認できるのでわかりやすいでですね。

以下の例は拒否されたトピックに登録した投資相談を行ってみた例です。最終応答が拒否されていることがわかります。

ちなみに応答メッセージの内容はガードレールのオプション設定で変更できます。

次に機密情報フィルターに設定したメールアドレスをダミーで生成してもらったところ、値がマスクされることを確認しました。

デプロイ

作成、編集したガードレールはまず「作業中のドラフト」という特別なバージョンに変更が反映されます。実際の環境で使用するには「バージョン」を作成し、ドラフトの内容をデプロイする必要があります。エージェントに近い管理方法です (エイリアスは存在しません)。

API からの利用

API での利用法自体は以下のドキュメントに記載されています。

モデル呼び出しの利用

boto3 では v1.34.90 以上で Guardrail が使用できることを確認しています。

アプリケーションから Guradrail を使用するには InvokeModel またはInvokeModelWithResponseStream リクエストを行うときにガードレールの ID とバージョンを指定します。

response = client.invoke_model_with_response_stream(
    body=body, 
    modelId="anthropic.claude-3-sonnet-20240229-v1:0",
    guardrailIdentifier="2hogeexample", # ガードレールID を指定
    guardrailVersion="1"
)

サンプルコード (同期モード):

app.py

import json
import boto3

def main():
    client = boto3.client('bedrock-runtime')

    body = json.dumps({
       "anthropic_version": "bedrock-2023-05-31",
       "max_tokens": 1024,
       "messages": [
           {
               "role": "user",
               "content": [{"type": "text", "text": "株式投資でFIREする方法を教えて"}]
           }
       ]
    })

    response = client.invoke_model_with_response_stream(
        body=body, 
        modelId="anthropic.claude-3-sonnet-20240229-v1:0",
        guardrailIdentifier="2hogeexample", # ガードレールID を指定
        guardrailVersion="1"
    )

    for event in response.get("body"):
       chunk = json.loads(event["chunk"]["bytes"])
       if chunk['type'] == 'content_block_delta' and chunk['delta']['type'] == 'text_delta':
           print(chunk['delta']['text'], end="")

if __name__ == "__main__":
    main()

実行結果 (同期モード)

$ python3 app.py
Sorry, the model cannot answer this question.

InvokeModelWithResponseStream API を使用したストリーミングレスポンスに対して、ガードレールを適用する場合、同期モードと非同期モードの 2 つの動作モードがあります。

デフォルトの同期モードでは、応答データをバッファリングしてポリシーを適用するため、レスポンスが遅れます。その代わりスキャンの精度が高くなります。上記のコード例では最終的な出力のみが表示されました。

非同期モードではレスポンスのチャンクがすぐに応答される代わりに、検出の精度が低下します。非同期モードを使用するにはリクエストの Body に以下のパラメーターを含めます。

{
    "amazon-bedrock-guardrailConfig": {
        "streamProcessingMode": "ASYNCHRONOUS"
    }
}

サンプルコード (非同期モード):

app.py

import json
import boto3

def main():
    client = boto3.client('bedrock-runtime')

    body = json.dumps({
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 1024,
        "messages": [
            {
                "role": "user",
                "content": [{"type": "text", "text": "株式投資でFIREする方法を教えて"}]
            }
        ],
        "amazon-bedrock-guardrailConfig": {
            "streamProcessingMode": "ASYNCHRONOUS"
        }
    })

    response = client.invoke_model_with_response_stream(
        body=body, 
        modelId="anthropic.claude-3-sonnet-20240229-v1:0",
        guardrailIdentifier="2hogeexample", # ガードレールID を指定
        guardrailVersion="1"
    )

    for event in response.get("body"):
       chunk = json.loads(event["chunk"]["bytes"])
       if chunk['type'] == 'content_block_delta' and chunk['delta']['type'] == 'text_delta':
           print(chunk['delta']['text'], end="")

if __name__ == "__main__":
    main()

実行結果 (非同期モード)

$ python3 app.py
株式投資でFIREを達成するには、長期的な視点とリスク管理が重要です。以下のステップを参考にしてください。

1. 収入の一部を確実に節約・投資する
まず収入の20-30%程度を投資に回せるように生活コストを抑える必要があります。投資額が多ければ早期リタイアに近づきます。

2. リスク許容度を理解し、ポートフォリオを構築する
100%株式に投資するのは高リスクです。債券、現金などを組み合わせ、自分のリスク許容度に合ったポートフォリオを作ります。

3.低コストの指数型投資信託などを選ぶ 
売買手数料や信託報酬の低い商品を選び、コストを最小限に抑えましょう。

4. 分散投資を心がける
複数の企業、業種、国に投資を分散させ、リスクを分散させます。

5. 長期的に保有し、積立投資を続ける
株価の変動に惑わされずに長期保有します。給与から定期的に積立投資を続けましょう。

6. マインドセットを変える
ライフスタイルの見直しや、FIRE後の収入源も検討して準備する必要があります。

一朝一夕にはいきませんが、規律ある投資を継続すればFIREに近づけます。自分に合った戦略を立ててみてく

非同期モードでは、応答がブロックされませんでした。日本語だと更に精度が落ちてしまうという可能性はありそうです。

エージェントでの利用

エージェントから利用する際は、エージェントの作成時や更新時にガードレール ID とバージョンを関連付ける必要があるようです。

boto3 v1.34.90 の時点ではまだ未対応で、コンソール上もまだガードレールの関連付けはできないようです。サポートされるのが待ち遠しいですね。

料金

ポリシー毎に料金設定が異なります。
2024/4/30 時点のバージニア北部リージョンの料金設定は以下のとおりです。

ポリシー	1,000 テキスト単位の料金
コンテンツフィルター	$0.75
拒否されたトピック	$1
ワードフィルター	Free
機密情報フィルター (PII)	$0.10
機密情報フィルター (正規表現)	Free

ワードフィルターの正規表現による機密情報フィルターは無料で利用できます
1 テキスト単位あたり最大 1,000 文字を含めることができます
- 1,000 文字以下は 1 テキスト単位として計算されます
- 入力が 5,600 文字の場合、6 テキスト単位分の料金が発生します

以上です。
参考になれば幸いです。