Dify のコンテンツモデレーションで Bedrock Guardrails を使う

Dify の「コンテンツモデレーション」を使うとチャット中の不適切な入出力をブロックできる．コンテンツモデレーションでは3種類のプロバイダがサポートされていて「キーワード」と「OpenAI モデレーション」は以下の記事にまとめた．今回はもう一つの「API 拡張機能」を試す❗️

kakakakakku.hatenablog.com

API 仕様

コンテンツモデレーションで API 拡張機能を使う場合は API 仕様に沿ったレスポンスを返す必要がある．詳しくはドキュメントにまとまっている．

legacy-docs.dify.ai

docs.dify.ai

今回試したアーキテクチャ

今回は AWS Lambda function URLs を使って AWS Lambda 関数を直接 API として公開して，コンテンツモデレーション部分は Amazon Bedrock Guardrails を使うアーキテクチャにした．AWS Lambda function URLs はあくまで簡易的に API を実現するために選択していて，Amazon API Gateway + AWS Lambda や ALB + Amazon ECS という組み合わせもあり得る👌

Amazon Bedrock Guardrails

まずは Amazon Bedrock Guardrails を準備する．今回は Terraform の aws_bedrock_guardrail リソースを使ってデプロイする．Amazon Bedrock Guardrails の設定は Creating Responsible AI With Amazon Bedrock Guardrails ワークショップを参考に投資関連のアドバイスを拒否するようにした．

resource "aws_bedrock_guardrail" "sandbox" {
  name                      = "sandbox"
  blocked_input_messaging   = "申し訳ありませんが、モデルはこの質問に回答できません。"
  blocked_outputs_messaging = "申し訳ありませんが、モデルはこの質問に回答できません。"

  content_policy_config {
    filters_config {
      type            = "VIOLENCE"
      input_strength  = "MEDIUM"
      output_strength = "MEDIUM"
    }

    tier_config {
      tier_name = "CLASSIC"
    }
  }

  topic_policy_config {
    topics_config {
      name       = "Financial Advice"
      definition = "Discussions that involve providing guidance, recommendations, or suggestions related to managing, investing, or handling finances, investments, or assets."
      examples = [
        "Can you suggest some good stocks to invest in right now ?",
        "What's the best way to save for retirement ?",
        "Should I put my money in a high-risk investment ?",
        "How can I maximize my returns on investments ?",
        "Is it a good time to buy real estate ?"
      ]
      type = "DENY"
    }

    tier_config {
      tier_name = "CLASSIC"
    }
  }
}

resource "aws_bedrock_guardrail_version" "sandbox" {
  description   = "sandbox"
  guardrail_arn = aws_bedrock_guardrail.sandbox.guardrail_arn
  skip_destroy  = true
}

AWS Lambda function URLs

次は AWS Lambda 関数を実装して AWS Lambda function URLs として公開する．Python で実装したサンプルコードは後半に貼っておきつつ，以下にポイントをまとめておく📝

まず，コンテンツモデレーションで API 拡張機能を設定するときに「API キー」を設定する必要がある．API キーは Dify から Authorization: Bearer {api_key} という形式で送信されるため，異なる場合は HTTP 401 を返すようにした．今回は許可する API キーを AWS Lambda 関数の環境変数 API_KEY に設定している．

headers = event.get('headers', {})
authorization = headers.get('authorization') or headers.get('Authorization')

if not authorization:
    return response401()

if not authorization.startswith('Bearer '):
    return response401()

token = authorization.replace('Bearer ', '').strip()
if token != API_KEY:
    return response401()

また API 拡張機能の API エンドポイントを設定するときに point=ping というパラメータが送信される．そのときに result=pong を返す必要がある．

if point == 'ping':
    return response200({'result': 'pong'})

最後に Boto3 で apply_guardrail() を呼び出して Amazon Bedrock Guardrails に問い合わせをしている．guardrailIdentifier と guardrailVersion は AWS Lambda 関数の環境変数 GUARDRAIL_ID と GUARDRAIL_VERSION に設定している．

boto3.amazonaws.com

try:
    guardrail_response = bedrock_runtime.apply_guardrail(
        guardrailIdentifier=GUARDRAIL_ID,
        guardrailVersion=GUARDRAIL_VERSION,
        source='INPUT',
        content=[
            {
                'text': {
                    'text': user_input,
                }
            }
        ],
    )

👾 app.py

import json
import os

import boto3

bedrock_runtime = boto3.client('bedrock-runtime')


API_KEY = os.environ.get('API_KEY')
GUARDRAIL_ID = os.environ.get('GUARDRAIL_ID')
GUARDRAIL_VERSION = os.environ.get('GUARDRAIL_VERSION')


def response200(body):
    return {
        'statusCode': 200,
        'body': json.dumps(body),
    }


def response401():
    return {
        'statusCode': 401,
        'body': json.dumps(
            {
                'error': 'Unauthorized',
            }
        ),
    }


def response500(message):
    return {
        'statusCode': 500,
        'body': json.dumps(
            {
                'error': message,
            }
        ),
    }


def lambda_handler(event, context):
    #
    # Check Authorization Header
    #
    headers = event.get('headers', {})
    authorization = headers.get('authorization') or headers.get('Authorization')

    if not authorization:
        return response401()

    if not authorization.startswith('Bearer '):
        return response401()

    token = authorization.replace('Bearer ', '').strip()
    if token != API_KEY:
        return response401()

    #
    # Decode Body
    #
    raw_body = event.get('body')
    body = json.loads(raw_body)

    point = body.get('point', {})
    if point == 'ping':
        return response200({'result': 'pong'})

    params = body.get('params', {})
    user_input = params.get('query', '')

    #
    # Apply Bedrock Guardrails
    #
    try:
        guardrail_response = bedrock_runtime.apply_guardrail(
            guardrailIdentifier=GUARDRAIL_ID,
            guardrailVersion=GUARDRAIL_VERSION,
            source='INPUT',
            content=[
                {
                    'text': {
                        'text': user_input,
                    }
                }
            ],
        )

        action = guardrail_response.get('action', 'NONE')
        intervened = action == 'GUARDRAIL_INTERVENED'

        if intervened:
            outputs = guardrail_response.get('outputs') or []
            preset = None

            if outputs and isinstance(outputs[0].get('text'), str):
                preset = outputs[0]['text']

            return response200(
                {
                    'flagged': True,
                    'action': 'direct_output',
                    'preset_response': preset,
                }
            )

        return response200(
            {
                'flagged': False,
                'action': 'direct_output',
            }
        )
    except Exception as e:
        return response500(str(e))

ちなみに動作確認をするときに Dify から AWS Lambda function URLs に送信されるイベントを取得したら以下のような感じだった．

{
    "version": "2.0",
    "routeKey": "$default",
    "rawPath": "/",
    "rawQueryString": "",
    "headers": {
        "content-length": "182",
        "x-amzn-tls-version": "TLSv1.3",
        "x-forwarded-proto": "https",
        "x-b3-sampled": "0",
        "x-forwarded-port": "443",
        "x-forwarded-for": "x.x.x.x",
        "accept": "*/*",
        "authorization": "Bearer xxxxx",
        "x-amzn-tls-cipher-suite": "TLS_AES_128_GCM_SHA256",
        "x-amzn-trace-id": "xxxxx",
        "x-b3-spanid": "xxxxx",
        "x-b3-traceid": "xxxxx",
        "traceparent": "xxxxx",
        "host": "xxxxx",
        "content-type": "application/json",
        "accept-encoding": "gzip, deflate, br, zstd",
        "user-agent": "python-httpx/0.27.2"
    },
    "requestContext": {
        "accountId": "anonymous",
        "apiId": "xxxxx",
        "domainName": "xxxxx",
        "domainPrefix": "xxxxx",
        "http": {
            "method": "POST",
            "path": "/",
            "protocol": "HTTP/1.1",
            "sourceIp": "x.x.x.x",
            "userAgent": "python-httpx/0.27.2"
        },
        "requestId": "xxxxx",
        "routeKey": "$default",
        "stage": "$default",
        "time": "04/Dec/2025:02:07:52 +0000",
        "timeEpoch": 1764814072481
    },
    "body": "{\"point\": \"app.moderation.input\", \"params\": {\"app_id\": \"xxxxx\", \"inputs\": {}, \"query\": \"Can you suggest some good stocks to invest in right now?\\n\\n\"}}",
    "isBase64Encoded": false
}