[Amazon Bedrock Flows] 冷蔵庫の食材と近所のスーパーのチラシから、レシピを提案するやつ作ってみました

// チラシ処理ノード - 統合Lambda（ノード名から処理対象を判定）
{
  name: 'FlyerProcessorNode',
  type: 'LambdaFunction',
  configuration: {
    lambdaFunction: {
      lambdaArn: imageProcessorLambda.functionArn,
    },
  },
  inputs: [
    {
      name: 'codeHookInput',
      type: 'String',
      expression: '$.data',
    },
  ],
  // ... 省略
},
// 冷蔵庫処理ノード - 同じLambdaを使用
{
  name: 'FridgeProcessorNode',
  type: 'LambdaFunction',
  configuration: {
    lambdaFunction: {
      lambdaArn: imageProcessorLambda.functionArn,  // 同一のLambda
    },
  },
  // ... 省略
},

def get_source_key_from_node_name(event: dict[str, Any]) -> str:
    """
    ノード名から処理対象のS3キーを判定する
    """
    node_name = event.get("node", {}).get("name", "")

    if "Fridge" in node_name:
        return "fridge.jpg"
    elif "Flyer" in node_name:
        return "flyer.jpg"
    else:
        # フォールバック: 環境変数から取得
        return os.environ.get("SOURCE_KEY", "fridge.jpg")

# boto3のタイムアウト設定（Claude Opusは処理に時間がかかるため延長）
BEDROCK_CONFIG = Config(
    read_timeout=600,  # 10分
    connect_timeout=60,
    retries={'max_attempts': 2}
)

def get_recipe_index_from_node_name(event: dict[str, Any]) -> int:
    """
    ノード名からrecipe_indexを判定する
    """
    node_name = event.get("node", {}).get("name", "")

    if "Dish1" in node_name:
        return 1
    elif "Dish2" in node_name:
        return 2
    elif "Dessert" in node_name:
        return 3
    else:
        return int(os.environ.get("RECIPE_INDEX", "1"))

def handler(event: dict[str, Any], context: Any) -> str:
    """
    複数の入力を受け取りレシピテキストを返す
    このLambdaは、3つの画像生成Lambdaとレシピ生成の完了を待機するために使用される。
    """
    # Bedrock Flowからの入力形式に対応
    if isinstance(event, dict) and "node" in event:
        node_inputs = event.get("node", {}).get("inputs", [])

        inputs_dict: dict[str, str] = {}
        for input_item in node_inputs:
            name = input_item.get("name", "")
            value = input_item.get("value", "")
            inputs_dict[name] = value

        recipe_text = inputs_dict.get("recipe_text", "")
        # image1_path, image2_path, image3_pathも受け取るが、
        # 待機のみが目的なので値は使用しない

    return str(recipe_text)

チラシ画像（flyer.jpg）がBedrockの5MB制限を超えています：
image exceeds 5 MB maximum: 6591084 bytes > 5242880 bytes

- S3から読み込んだサイズ: 4.9MB
- Bedrockに送信時のサイズ: 6.59MB（Base64エンコードで増加）

def compress_image_if_needed(image_path: Path, max_size_mb: float = 3.5) -> bytes:
    """
    画像が指定サイズを超える場合は圧縮する
    """
    max_size_bytes = int(max_size_mb * 1024 * 1024)
    original_size = image_path.stat().st_size

    if original_size <= max_size_bytes:
        return image_path.read_bytes()  # 圧縮不要

    # 段階的に圧縮を試行
    quality = 85
    max_dimension = 2000

    while quality >= 30:
        if max(img.size) > max_dimension:
            ratio = max_dimension / max(img.size)
            new_size = (int(img.size[0] * ratio), int(img.size[1] * ratio))
            resized_img = img.resize(new_size, Image.Resampling.LANCZOS)

        buffer = io.BytesIO()
        resized_img.save(buffer, format="JPEG", quality=quality, optimize=True)
        compressed_data = buffer.getvalue()

        if len(compressed_data) <= max_size_bytes:
            return compressed_data

        quality -= 10        # 85 → 75 → 65 → ...
        max_dimension -= 200  # 2000 → 1800 → 1600 → ...

cd cdk
npm install
cdk deploy

aws cloudformation describe-stacks \
  --stack-name FridgeFlyerStack \
  --query "Stacks[0].Outputs" \
  --output table

cd recipe
python3 generate_recipe.py

============================================================
Fridge Flyer - レシピ生成スクリプト
============================================================
画像をS3にアップロード中...
  - flyer.jpg -> s3://fridge-flyer-XXXX/flyer.jpg
  - fridge.jpg -> s3://fridge-flyer-XXXX/fridge.jpg
Bedrock Flowを実行中...
  （画像分析とレシピ生成に数分かかります。お待ちください...）
  - 出力を受信しました
  - Flow完了
  - レスポンス取得完了（1906文字）
HTMLを生成中...
  - HTMLを保存: output/index.html
============================================================
完了！
============================================================