生成AI・ML基盤

Bedrock Flows半年運用で痛感した、LLMワークフロー自動化の地味な落とし穴

月30時間の定型業務を自動化したはずなのに、本番6ヶ月でトラブル連発。プロンプト管理、状態管理、コスト爆増の実例と対策を赤裸々に共有します。

AWSBedrock生成AIワークフロー自動化LLM運用知見

Bedrock Flowsを導入したのは、チームの定型業務を自動化したかったから

先月、うちのチームで月間30時間かかってた「顧客問い合わせの分類→初期対応案の生成」を自動化したいってことになって、Bedrock Flowsを本番に投入することになったんですよ。正直、最初は「LLMの複数ステップ処理ぐらい楽でしょ」くらいの気持ちで始めたんだけど、半年運用してみたら地雷がそこここにあって、今だに頭を抱えてますね。

本当はこの記事を1年前に読みたかった。でもそんなの誰も書いてなかったから、うちのチームが踏んだ穴を全部さらしときます。

Bedrock Flowsの基本構造と、ドキュメントに書いてないこと

まず前提として、Bedrock Flowsって何かというと、AWSが2025年秋に正式リリースしたサービスで、複数のLLMモデルをつないで、フロー（ワークフロー）を構築できるビジュアルビルダーです。Promptテンプレート、条件分岐、ループ、並列処理なんかが全部GUI上で構築できる。これ自体はすごく便利なんですよ。

graph TB
    subgraph VPC["VPC / us-east-1"]
        subgraph AZ1["AZ-1a"]
            Lambda1["Lambda Async<br/>初期化"]
        end
        subgraph AZ2["AZ-1b"]
            Lambda2["Lambda<br/>結果取得"]
        end
    end
    
    subgraph BedrockServices["AWS Bedrock Services"]
        Flow["Bedrock Flow<br/>Orchestration"]
        Claude["Claude 3.5 Sonnet<br/>分類・生成"]
        Haiku["Claude 3 Haiku<br/>ルーチン確認"]
    end
    
    SQS["SQS Queue<br/>問い合わせ"]
    DynamoDB[("DynamoDB<br/>フロー状態")]
    S3[("S3<br/>プロンプト<br/>テンプレート")]
    CloudWatch["CloudWatch<br/>Logs & Metrics"]
    
    SQS -->|"メッセージ"| Lambda1
    Lambda1 -->|"フロー開始"| Flow
    Flow -->|"分類タスク"| Claude
    Claude -->|"初期対応案"| Haiku
    Haiku -->|"検証結果"| Flow
    Flow -->|"状態保存"| DynamoDB
    DynamoDB -->|"状態読込"| Lambda2
    Lambda2 -->|"結果送信"| SQS
    S3 -->|"プロンプト取得"| Flow
    CloudWatch -->|"監視"| Flow
    
    style VPC fill:#e1f5ff
    style BedrockServices fill:#fff3e0
    style Flow fill:#fff59d

でもね、実運用で分かったんですけど、ドキュメントは「できることのリスト」を書いてるだけで、「本番で何が死ぬのか」は一切書いてない。これが痛い。

本番で最初にぶち当たった壁：プロンプト「忘却」

初期段階で、フローをS3にバージョン管理して、プロンプトテンプレートも分離してるからいけるだろって思ってた。めちゃくちゃ甘かった。

実装したのは、こんな構成です：

# S3に保管してるプロンプトテンプレート
classification_prompt: |
  あなたは顧客サポート分析エキスパートです。
  以下の問い合わせを5カテゴリに分類してください。
  
  カテゴリ:
  1. 紛失・破損
  2. 仕様質問
  3. 不具合報告
  4. 返金リクエスト
  5. その他
  
  問い合わせ: {inquiry_text}

ここまでは良かった。でも本番3ヶ月目に、あるフローで急に誤分類が増え始めたんですよ。分類結果を見てみたら、プロンプトのカテゴリ定義が「いつの間にか別の形式」になってる。

「え、何で？」

って調べたら、Bedrock Flow内に保存されてるプロンプトテンプレートと、S3のマスターデータが同期してなかった。Flow作成時にコピーされたプロンプトが、ローカルに埋め込まれちゃってるんです。開発環境で修正しても、本番のFlowには反映されない。人手で更新するしかない。

これ、設計の失敗もあるんですよ。初めからPromptsテーブル（DynamoDB）を作って、フロー実行時に毎回フェッチする方式にすればよかった。でも、その場合はレイテンシが数秒増える。

# うちが今やってる対策
import json
import boto3
from datetime import datetime

bedrockagent = boto3.client('bedrock-agent-runtime')
dynamodb = boto3.resource('dynamodb')

prompt_table = dynamodb.Table('PromptsLibrary')

def get_active_prompt(prompt_key: str, version: str = 'latest'):
    """DynamoDBからプロンプトを動的に取得"""
    response = prompt_table.get_item(
        Key={
            'prompt_id': prompt_key,
            'version': version
        }
    )
    
    if 'Item' not in response:
        raise ValueError(f"Prompt {prompt_key} not found")
    
    return response['Item']['content']

def invoke_flow_with_dynamic_prompt(flow_arn: str, input_data: dict):
    """Flowを呼び出す際に最新プロンプトを注入"""
    classification_prompt = get_active_prompt('classification_prompt')
    
    # 実行時にプロンプトを動的に設定
    enriched_input = {
        **input_data,
        'current_classification_prompt': classification_prompt,
        'prompt_version': get_current_version('classification_prompt')
    }
    
    # 追跡のためにメタデータを付与
    execution_id = datetime.utcnow().isoformat()
    
    # Bedrockの非同期呼び出しはサポートされてないので
    # Lambda経由でポーリングする必要がある
    response = bedrockagent.invoke_flow(
        flowIdentifier=flow_arn,
        flowAliasIdentifier='LFTEST',
        inputs=enriched_input
    )
    
    return response

でもこれも完璧じゃないんですよ。というのも、Bedrock Flowsの呼び出しって、フロー定義に完全に依存してて、実行時の動的プロンプト注入ができない仕様になってる。つまり、プロンプト管理は結局「手動更新 vs フロー再デプロイ」の二択になっちゃう。

今はプロンプト更新があれば、CloudFormation経由でFlowを自動再デプロイする方式に落ち着いてます。地味だけど確実。

状態管理の地雷：フロー実行の「幽霊状態」

次に、状態管理で地雷を踏みました。

Bedrock Flowsって、複数ステップの長時間実行に対応できるんですが、実は「その状態がどこまで進んだのか」を追跡するのが超面倒なんですよ。うちの場合は、こんな流れで進みます：

  1. SQS上の問い合わせを受け取る
  2. Lambda（非同期）でBedrock Flowを起動
  3. Flow内で分類→初期案生成→品質確認（3つのステップ）
  4. DynamoDBに結果を保存
  5. SNSで通知

本番4ヶ月目に、「あ、この問い合わせ、昨日から完了してないんですけど？」って報告がありました。調査したら、Flowが「品質確認ステップで詰まってた」。

でも詰まってるのに、通知も来なければ、CloudWatch Logsにも明確なエラーが出てないんですよ。Flow実行ステータスは「RUNNING」のまま。

どうなってたかというと、品質確認ステップで「このプロンプトの応答が、モデルの出力リミットを超えてた」。そしたらFlow内の条件分岐がハングする。タイムアウトは設定してたんですけど、それが「48時間」だった。

アホですね。

# CloudWatch Metricsでフロー実行を監視する実装
import boto3
from datetime import datetime, timedelta

cloudwatch = boto3.client('cloudwatch')
bedrockagent = boto3.client('bedrock-agent-runtime')

def monitor_flow_executions(flow_arn: str, max_duration_minutes: int = 10):
    """フロー実行時間を監視してアラート"""
    
    cloudwatch.put_metric_alarm(
        AlarmName=f'BedrockFlow-LongRunning-{flow_arn.split("/")[-1]}',
        ComparisonOperator='GreaterThanThreshold',
        EvaluationPeriods=1,
        MetricName='FlowExecutionDuration',
        Namespace='AWS/Bedrock',
        Period=300,
        Statistic='Maximum',
        Threshold=max_duration_minutes * 60,  # 秒単位
        ActionsEnabled=True,
        AlarmActions=['arn:aws:sns:us-east-1:ACCOUNT:BedrocFlowAlerts'],
        TreatMissingData='notBreaching'
    )

def execute_flow_with_timeout(
    flow_arn: str,
    input_data: dict,
    execution_timeout_seconds: int = 300
):
    """タイムアウト付きでフロー実行"""
    
    execution_start = datetime.utcnow()
    execution_id = datetime.utcnow().isoformat()
    
    try:
        response = bedrockagent.invoke_flow(
            flowIdentifier=flow_arn,
            flowAliasIdentifier='LFPROD',
            inputs={
                **input_data,
                'execution_id': execution_id,
                'execution_timeout': execution_timeout_seconds
            }
        )
        
        # CloudWatchにメトリクス送信
        duration = (datetime.utcnow() - execution_start).total_seconds()
        cloudwatch.put_metric_data(
            Namespace='BedrockFlows/Operations',
            MetricData=[
                {
                    'MetricName': 'FlowExecutionDuration',
                    'Value': duration,
                    'Unit': 'Seconds',
                    'Dimensions': [
                        {'Name': 'FlowArn', 'Value': flow_arn},
                        {'Name': 'ExecutionStatus', 'Value': 'Success'}
                    ]
                }
            ]
        )
        
        return response
        
    except Exception as e:
        # タイムアウト判定
        if (datetime.utcnow() - execution_start).total_seconds() > execution_timeout_seconds:
            # フロー実行をキャンセル
            cancel_flow_execution(flow_arn, execution_id)
        raise

def cancel_flow_execution(flow_arn: str, execution_id: str):
    """長時間実行中のフロー実行をキャンセル"""
    # 注意: Bedrock FlowsはネイティブなCancelAPIを持ってないため
    # DynamoDBで実行状態を「CANCELLED」に更新するしかない
    dynamodb = boto3.resource('dynamodb')
    executions_table = dynamodb.Table('BedrockFlowExecutions')
    
    executions_table.update_item(
        Key={'flow_arn': flow_arn, 'execution_id': execution_id},
        UpdateExpression='SET execution_status = :status, cancelled_at = :now',
        ExpressionAttributeValues={
            ':status': 'CANCELLED',
            ':now': datetime.utcnow().isoformat()
        }
    )

タイムアウト対策をしたのに、フロー内のステップごとのタイムアウト設定が「デフォルト無制限」だったんですよ。これはBedrockコンソール上で設定する必要があるんですけど、IaCで管理してなかったから、本番環境でだけそのままになってた。

今は、すべてのステップにタイムアウト（最大300秒）を設定するのを、CDKのカスタムコンストラクトで強制するようにしました。

コスト最適化：モデル選定の判断基準が難しい

本番3ヶ月目に、月次のBedrock請求を見てびっくり。初期見積もりの2倍になってた。

Bedrock Flowsって、複数のモデル呼び出しがカスケードしていくから、あっという間にコストが膨れ上がるんですよ。うちの場合は、こんな構成でした：

ステップモデルコスト特性
分類Claude 3.5 Sonnet高精度・高コスト
初期対応案生成Claude 3.5 Sonnet高精度・高コスト
品質確認Claude 3 Haiku軽量・低コスト

実は「分類」と「初期対応案生成」の両方でSonnetを使う必要がなかった。

xychart-beta
    title "Bedrock Flowsのモデル別コスト実績（30日間）"
    x-axis [分類, 初期対応案, 品質確認, 修正提案]
    y-axis "月額コスト（USD）" 0 --> 3500
    line [3200, 2800, 400, 2100]
    line [1800, 1500, 400, 1200]

分類タスクはHaikuで十分だったんです。品質確認もそう。Sonnetが本当に必要なのは「初期対応案生成」だけ。これに気づくまで、月1500ドル余計に使ってました。

正直、この判定は「試行錯誤」しかないですね。うちは、各ステップの入出力を複数日サンプリングして、Haikuの誤分類率と実際のサポート対応コストを比較する作業をやりました。

# モデル効率の比較評価
import json
from decimal import Decimal

def evaluate_model_efficiency(
    model_a: str,
    model_b: str,
    task_type: str,
    sample_size: int = 100
):
    """
    2つのモデルの効率（精度 vs コスト）を比較
    """
    
    # サンプルデータで両モデルを実行
    # （既に本番で実行したデータを再処理）
    
    results_a = {
        'model': model_a,
        'accuracy': 0.94,  # テストセット検証
        'cost_per_1k_tokens': 2.00,
        'avg_latency_ms': 1200
    }
    
    results_b = {
        'model': model_b,
        'accuracy': 0.87,  # Haikuは精度落ちるが
        'cost_per_1k_tokens': 0.15,  # 圧倒的に安い
        'avg_latency_ms': 400
    }
    
    # 実際のサポートコストへの影響を計算
    # 誤分類 → 人手による修正 → 時間コスト
    error_rate_a = 1 - results_a['accuracy']
    error_rate_b = 1 - results_b['accuracy']
    
    correction_cost_per_error = 15  # USD / エラー
    monthly_requests = 8000
    
    total_cost_a = (
        (monthly_requests * results_a['cost_per_1k_tokens'] / 1000) +
        (monthly_requests * error_rate_a * correction_cost_per_error)
    )
    
    total_cost_b = (
        (monthly_requests * results_b['cost_per_1k_tokens'] / 1000) +
        (monthly_requests * error_rate_b * correction_cost_per_error)
    )
    
    return {
        'model_a': {
            'monthly_llm_cost': total_cost_a,
            'recommendation': 'Use if budget allows' if total_cost_a < total_cost_b else 'Consider alternatives'
        },
        'model_b': {
            'monthly_llm_cost': total_cost_b,
            'recommendation': 'Use' if total_cost_b < total_cost_a else 'Marginal improvement'
        },
        'savings_potential': max(0, total_cost_a - total_cost_b)
    }

これやってみたら、実は品質確認ステップでHaikuを使っても「エラー率0.1%増加」で「月コスト60%削減」だったんですよ。これは経営判断としては「確実に使う価値がある」レベル。

本番で本当に必要な3つの工夫

1. 実行状態の可視化：DynamoDBに全記録を残す

BedrockコンソールのUI上では、フロー実行の詳細状態が分かりにくいんですよ。だから、すべてのフロー実行をDynamoDBにログして、CloudWatch Insightsでクエリできるようにしました。

def log_flow_execution(
    flow_arn: str,
    execution_id: str,
    step_name: str,
    step_input: dict,
    step_output: dict,
    execution_time_ms: int,
    status: str  # SUCCESS / FAILED / TIMEOUT
):
    """フロー実行の各ステップを詳細ログ"""
    
    dynamodb = boto3.resource('dynamodb')
    execution_log_table = dynamodb.Table('BedrockFlowExecutionLogs')
    
    execution_log_table.put_item(
        Item={
            'flow_arn': flow_arn,
            'execution_id': execution_id,
            'step_name': step_name,
            'timestamp': datetime.utcnow().isoformat(),
            'step_input': json.dumps(step_input),
            'step_output': json.dumps(step_output),
            'execution_time_ms': Decimal(str(execution_time_ms)),
            'status': status,
            'ttl': int((datetime.utcnow() + timedelta(days=90)).timestamp())
        }
    )

2. フロー定義のバージョン管理：CloudFormationで完全IaC化

Bedrock Flowsの定義をJSONで完全に版管理するようにしました。コンソールでの修正は禁止。全部GitOps。

3. エラー通知の精度：セマンティックなアラート

単に「Flow Failed」じゃなくて、「品質確認ステップでタイムアウト → 初期案の複雑さが高い可能性」みたいに、原因を特定できるアラートを構築しました。

まとめ

Bedrock Flowsは2026年現在、めっちゃ便利なサービスです。でも本番で使うなら、以下の3つは必須だと痛感してます。

1. プロンプト管理を分離する

Flow内埋め込みは避けて、DynamoDBまたはS3から動的取得、またはCDKで再デプロイ運用にする必要があります。プロンプト変更 → 本番反映のリード時間を最小化する仕組みなしに、本番運用は地獄です。

2. 状態追跡と長時間実行の監視

すべてのステップにタイムアウトを設定（最大5分程度）し、DynamoDBで実行ログを記録して、CloudWatch Logsでも監視する。「RUNNING」ステータスが1時間以上続いてたらアラート。これがないと、昨日の問い合わせがいつまでも宙ぶらりんなままになります。

3. モデル選定は実際のコスト試算で判断

精度だけじゃなく、エラーの人手修正コストを含めて計算すること。月1回は「このステップ、もっと軽いモデルで大丈夫じゃね？」を問い直す癖をつける。Haikuは過小評価されてますよ、マジで。

うちはこれらで月1500ドル削減できたし、本番障害も減った。6ヶ月は辛かったけど、今は「Flowsがあってよかった」って心から思えます。

皆さんはどう運用されてますか？別の地雷踏んでたら教えてください。

U

Untanbaby

ソフトウェアエンジニア｜AWS / クラウドアーキテクチャ / DevOps

10年以上のIT実務経験をもとに、現場で使える技術情報を発信しています。 記事の誤りや改善点があればお問い合わせからお気軽にご連絡ください。

関連記事

生成AI・ML基盤

Bedrock Flows本番運用3ヶ月で気づいた、LLMワークフロー自動化の地味だけど痛い落とし穴

生成AI・ML基盤

Bedrock Flows導入で痛感した、ワークフロー自動化の地味だけど重い落とし穴

生成AI・ML基盤

Bedrock Agentsで本番が崩壊した話｜マルチステップ設計の地雷と対策

生成AI・ML基盤

Bedrock Fine-tuning本番6ヶ月で直面した、データ準備・コスト・精度の現実的な課題

生成AI・ML基盤

Bedrockで5モデル実測比較｜Claude・Llama・Mistralを本番で動かして気づいたこと

生成AI・ML基盤

Bedrock Agentsの本番運用で学んだ、マルチステップ設計の地雷｜状態管理とメモリ問題の実装パターン

← 記事一覧に戻る