Step FunctionsでAIワークフローを組む ― 並列要約・リトライ・人間承認を2026年の機能で作り直す

LLMに複数の処理を順番にやらせるとき、最初はたいてい1本のスクリプトに for ループを書きます。100件の文書を要約する、下書きを作ってレビューさせて直させる。件数が少ないうちはそれで回ります。ところが件数が増えると、途中でスロットリングに当たって止まり、どこまで終わったのか分からないまま最初からやり直す、ということが起きます。リトライの待ち時間、途中再開、誰がいつ承認したかの記録。本来やりたい処理とは関係のないコードが、だんだん本体より長くなっていきます。

2025年11月の「AI劣化でデスマーチ」では、長い処理をAIに一気にやらせず「小さく刻んで並列に回す」ほうがよいと書きました。そのときは実装例としてGitHub ActionsのMatrix戦略を挙げています。この記事は、同じ考え方をAWSのマネージドサービスで組むとどうなるか、という続きです。

このネタは2026年7月にメモしていたものですが、書こうとして調べ直すと、その後に前提がかなり変わっていました。Step FunctionsからBedrock AgentCoreのエージェントを直接呼べるようになり、Lambdaには「durable functions」という競合しそうな仕組みも入っています。そこで、2026年10月時点の公式ドキュメントで確認できた内容に合わせて組み直しました。

先に断っておくと、記事中のステートマシン定義（ASL）は公式ドキュメントの仕様に沿って書き、JSONとして正しいことは確認していますが、私のAWSアカウントにデプロイして動かした結果ではありません。使うときはWorkflow Studioに貼り付けるか、TestState APIで1ステートずつ確かめてから進めてください。

2026年までに何が変わったか

以前のStep Functionsの記事やサンプルを見て「JSONPathの ResultPath 地獄がつらい」という印象を持っている人は、まずここを読み直してほしいところです。AIワークフローに関係する変更を、公開日の順に並べます。

この記事のサンプルはすべてJSONataで書いています。既存のJSONPathの定義はそのまま動き、ステートごとに混在もできるので、切り替えは新しく作るものからで構いません。

なぜループではなくワークフローエンジンなのか

スクリプトのループとStep Functionsの差は、処理の中身よりも「失敗したとき」に出ます。

• どこまで終わったかが残る: 実行履歴に各ステップの入出力が残る。Standardワークフローなら終了後90日間コンソールで見られる
• 途中から再開できる: 失敗した実行は、終了から14日以内なら失敗したステップから「redrive」できる。成功済みの要約をもう一度LLMに投げなくて済む
• リトライが宣言で済む: 待ち時間の倍率、上限、ジッターを定義に書くだけで、指数バックオフのコードを書かなくていい
• 人を待てる: 承認待ちで最長1年止めておける。止まっている間は状態遷移が起きないので課金もされない

「小さく刻んで並列に回す」と相性がいいのは、2つ目です。100件のうち3件だけ失敗したとき、ループなら3件を探して流し直す仕組みを自分で作ることになります。Distributed Mapなら失敗した子実行だけをredriveできます。

Lambda durable functionsとどちらを使うか

2025年12月に出たLambda durable functionsは、Lambdaのコード（TypeScript/JavaScriptかPython）の中にステップと待機を書くと、そこまでの進み具合がチェックポイントとして保存される仕組みです。失敗しても最初からではなく続きから再開でき、待機中はコンピューティングの料金がかかりません。かなり雑に言うと「Step Functionsの再開とリトライを、普通のコードの書き方で使えるようにしたもの」です。

AWSのドキュメントは使い分けを次のように整理しています。私もこの線引きに賛成です。

この記事で扱う「S3の記事を読んで、Bedrockで要約して、S3に書いて、人に承認を求める」は、サービスをまたぐ処理がほとんどで、Lambdaのコードがほぼ出てきません。こういう形はStep Functionsのほうが向いています。

AIワークフローの基本パターン3つ

図：AIワークフローの3パターン

A. 直列チェーン：生成→レビュー→修正

LLMに下書きを作らせ、別のプロンプトでレビューさせ、指摘があれば直させる形です。ここで一番大事なのは、修正の回数に上限を置くことです。LLM同士で「直した」「まだ駄目」を延々と繰り返すと、トークンだけが減っていきます。下の定義では、修正が2回を超えたら人間に回します。

Bedrockの呼び出しには、Step Functionsの最適化統合 arn:aws:states:::bedrock:invokeModel を使います。Lambdaを挟まずにモデルを呼べ、レスポンスの本文は Body に入って返ってきます。Body の中身はモデルごとの形式で、ここではAnthropicのClaudeを想定しています。

{
  "QueryLanguage": "JSONata",
  "StartAt": "Draft",
  "States": {
    "Draft": {
      "Type": "Task",
      "Resource": "arn:aws:states:::bedrock:invokeModel",
      "Arguments": {
        "ModelId": "{% $states.input.modelId %}",
        "Body": {
          "anthropic_version": "bedrock-2023-05-31",
          "max_tokens": 2000,
          "messages": [{ "role": "user", "content": "{% $states.input.request %}" }]
        }
      },
      "Assign": {
        "modelId": "{% $states.input.modelId %}",
        "draft": "{% $states.result.Body.content[0].text %}",
        "round": 0
      },
      "Next": "Review"
    },
    "Review": {
      "Type": "Task",
      "Resource": "arn:aws:states:::bedrock:invokeModel",
      "Arguments": {
        "ModelId": "{% $modelId %}",
        "Body": {
          "anthropic_version": "bedrock-2023-05-31",
          "max_tokens": 500,
          "messages": [{
            "role": "user",
            "content": "{% '次の文章をレビューし、{\"ok\": true/false, \"issues\": [...]} のJSONだけを返してください。\\n\\n' & $draft %}"
          }]
        }
      },
      "Assign": {
        "verdict": "{% $parse($states.result.Body.content[0].text) %}"
      },
      "Catch": [
        { "ErrorEquals": ["States.QueryEvaluationError"], "Next": "NeedsHuman" }
      ],
      "Next": "IsOk"
    },
    "IsOk": {
      "Type": "Choice",
      "Choices": [
        { "Condition": "{% $verdict.ok %}", "Next": "Done" },
        { "Condition": "{% $round >= 2 %}", "Next": "NeedsHuman" }
      ],
      "Default": "Revise"
    },
    "Revise": {
      "Type": "Task",
      "Resource": "arn:aws:states:::bedrock:invokeModel",
      "Arguments": {
        "ModelId": "{% $modelId %}",
        "Body": {
          "anthropic_version": "bedrock-2023-05-31",
          "max_tokens": 2000,
          "messages": [{
            "role": "user",
            "content": "{% '指摘に沿って修正してください。\\n指摘: ' & $string($verdict.issues) & '\\n\\n' & $draft %}"
          }]
        }
      },
      "Assign": {
        "draft": "{% $states.result.Body.content[0].text %}",
        "round": "{% $round + 1 %}"
      },
      "Next": "Review"
    },
    "Done": { "Type": "Pass", "Output": { "text": "{% $draft %}" }, "End": true },
    "NeedsHuman": { "Type": "Fail", "Error": "NeedsHuman", "Cause": "自動修正の上限に達したか、レビュー結果を解釈できませんでした" }
  }
}

変数（Assign）が使えるようになったおかげで、$draft や $round を状態をまたいで持ち回せます。JSONPath時代は、これを入力のJSONに継ぎ足しながら運ぶ必要がありました。

詰まりやすいのはレビュー結果の解釈です。「JSONだけを返して」と頼んでも、モデルが前後に説明文を付けてくることはあります。そのとき $parse が失敗して States.QueryEvaluationError になるので、Catch で人間に回すようにしています。本番では、レビューにはJSONで返すよう強制できる仕組み（ツール指定や構造化出力）を使うほうが確実です。

B. Distributed Map：同時実行数を絞った並列処理

大量の文書を1件ずつ独立して処理するなら、Distributed Mapを使います。S3上のファイルを読み、1件ごとに子ワークフローを起動して並列に処理する仕組みです。子実行は1つのMap Runで最大1万まで同時に動かせます。

ただし、LLMを呼ぶ場合に効いてくる上限はStep Functions側ではなく、Bedrockのクォータ（1分あたりのリクエスト数やトークン数）です。1万並列で投げても、ほとんどがスロットリングで弾かれてリトライに回るだけです。MaxConcurrency は、Bedrockのクォータから逆算して決めます。たとえば1件あたり平均5,000トークン、1回の処理に平均20秒かかり、使えるクォータが1分あたり10万トークンなら、1分間に処理できるのは20件で、同時に走らせてよいのは7件前後です。そこから少し余裕を見て5にする、という決め方になります。

もう1つの設定が ToleratedFailurePercentage です。1件でも失敗したらMap全体を失敗にするのか、1割までは許容して先に進むのか。要約のように「あとで失敗分だけやり直せばいい」処理なら、許容しておいたほうが運用は楽です。

C. 人間承認：タスクトークンで待つ

AIが作ったものを人が確認してから公開したい、という要件はほぼ必ず出てきます。Step Functionsでは、リソースの末尾に .waitForTaskToken を付けると、そのステートはタスクトークンが返ってくるまで止まります。トークンは $states.context.Task.Token で取り出して通知に載せ、承認する側が SendTaskSuccess（差し戻すなら SendTaskFailure）にトークンを渡すと先に進みます。

注意点が3つあります。

• Standardワークフローでしか使えない: Expressは最大5分で、.waitForTaskToken にも対応していない
• タイムアウトを必ず付ける: 付けないと最長1年待ち続ける。TimeoutSeconds で承認期限を決め、期限切れは差し戻しと同じ扱いにする
• トークンを返すのは同じAWSアカウントの主体: 別アカウントからは返せない。承認する人やシステムには states:SendTaskSuccess の権限が要る

ハンズオン：ブログ記事を一括要約して承認を待つ

3つのパターンのうちBとCを組み合わせて、「S3に置いた記事一覧を並列で要約し、結果をS3に書き出し、人の承認を待ってから公開処理に進む」ワークフローを書きます。このブログでも記事ごとに各媒体向けの紹介文を用意しているので、それを一括で下書きさせる用途を想定しています。

入力のJSONLは、1行に1記事です。

{"url": "https://example.com/post-1", "title": "記事タイトル1", "text": "本文..."}
{"url": "https://example.com/post-2", "title": "記事タイトル2", "text": "本文..."}

実行時の入力には、バケット名、JSONLのキー、モデルIDを渡します。モデルIDはリージョンや推論プロファイルによって変わるので、定義に埋め込まず入力で渡すようにしました。

{
  "bucket": "my-summary-bucket",
  "itemsKey": "input/2026-10/articles.jsonl",
  "modelId": "（Bedrockコンソールで確認した推論プロファイルID）"
}

ステートマシンの定義全体です。

{
  "Comment": "ブログ記事の一括要約：Distributed Mapで並列要約し、人間の承認を待ってから公開する",
  "QueryLanguage": "JSONata",
  "StartAt": "Init",
  "States": {
    "Init": {
      "Type": "Pass",
      "Assign": {
        "bucket": "{% $states.input.bucket %}",
        "modelId": "{% $states.input.modelId %}",
        "runId": "{% $states.context.Execution.Name %}"
      },
      "Next": "SummarizeAll"
    },
    "SummarizeAll": {
      "Type": "Map",
      "ItemReader": {
        "Resource": "arn:aws:states:::s3:getObject",
        "ReaderConfig": { "InputType": "JSONL" },
        "Arguments": {
          "Bucket": "{% $bucket %}",
          "Key": "{% $states.input.itemsKey %}"
        }
      },
      "ItemSelector": {
        "url": "{% $states.context.Map.Item.Value.url %}",
        "title": "{% $states.context.Map.Item.Value.title %}",
        "text": "{% $states.context.Map.Item.Value.text %}",
        "modelId": "{% $modelId %}"
      },
      "MaxConcurrency": 5,
      "ToleratedFailurePercentage": 10,
      "ItemProcessor": {
        "ProcessorConfig": { "Mode": "DISTRIBUTED", "ExecutionType": "EXPRESS" },
        "StartAt": "Summarize",
        "States": {
          "Summarize": {
            "Type": "Task",
            "Resource": "arn:aws:states:::bedrock:invokeModel",
            "Arguments": {
              "ModelId": "{% $states.input.modelId %}",
              "Body": {
                "anthropic_version": "bedrock-2023-05-31",
                "max_tokens": 800,
                "system": "あなたは技術ブログの編集者です。事実を足さずに要約してください。",
                "messages": [
                  {
                    "role": "user",
                    "content": "{% '次の記事を400字以内で要約してください。\\n\\nタイトル: ' & $states.input.title & '\\n\\n' & $states.input.text %}"
                  }
                ]
              }
            },
            "Retry": [
              {
                "ErrorEquals": ["Bedrock.ThrottlingException", "Bedrock.ServiceUnavailableException"],
                "IntervalSeconds": 2,
                "MaxAttempts": 6,
                "BackoffRate": 2,
                "MaxDelaySeconds": 60,
                "JitterStrategy": "FULL"
              }
            ],
            "Output": {
              "url": "{% $states.input.url %}",
              "summary": "{% $states.result.Body.content[0].text %}",
              "inputTokens": "{% $states.result.Body.usage.input_tokens %}",
              "outputTokens": "{% $states.result.Body.usage.output_tokens %}"
            },
            "End": true
          }
        }
      },
      "ResultWriter": {
        "WriterConfig": { "OutputType": "JSONL", "Transformation": "COMPACT" },
        "Resource": "arn:aws:states:::s3:putObject",
        "Arguments": {
          "Bucket": "{% $bucket %}",
          "Prefix": "{% 'summaries/' & $runId %}"
        }
      },
      "Next": "WaitForApproval"
    },
    "WaitForApproval": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish.waitForTaskToken",
      "Arguments": {
        "TopicArn": "arn:aws:sns:ap-northeast-1:123456789012:summary-review",
        "Subject": "Summary review request",
        "Message": "{% '要約結果: s3://' & $states.input.ResultWriterDetails.Bucket & '/' & $states.input.ResultWriterDetails.Key & '\\n承認する場合は、次のトークンで SendTaskSuccess を呼んでください。\\n' & $states.context.Task.Token %}"
      },
      "TimeoutSeconds": 259200,
      "Catch": [
        { "ErrorEquals": ["States.ALL"], "Next": "NotApproved" }
      ],
      "Next": "Publish"
    },
    "Publish": {
      "Type": "Pass",
      "Comment": "ここで公開処理（S3へのコピーやCMSのAPI呼び出し）を行う",
      "End": true
    },
    "NotApproved": {
      "Type": "Fail",
      "Error": "NotApproved",
      "Cause": "レビューで差し戻されたか、承認期限を過ぎました"
    }
  }
}

定義のポイント

子ワークフローはExpressにしています。1件の要約は数十秒で終わるので、5分の上限に収まります。Expressは状態遷移ではなくリクエスト数と実行時間で課金されるため、短い処理を大量に回すなら安くなります。親はStandardのままなので、承認待ちもredriveもできます。

リトライの最大待ち時間を計算しておきます。この設定だと待ち時間は2, 4, 8, 16, 32, 60秒（MaxDelaySeconds で頭打ち）で、ジッターを FULL にしているので実際はそれぞれ0秒からその値までのランダムになります。最悪でも合計約2分なので、Expressの5分に収まります。MaxAttempts を増やすときは、この合計が5分を超えないかを先に確認してください。超えると子実行が States.Timeout で落ちます。

エラー名は実行履歴で確かめます。ErrorEquals の文字列は大文字小文字まで完全一致です。最適化統合のエラーは Bedrock.ThrottlingException のように「サービス名.例外名」の形になりますが、初回は実行履歴の失敗イベントに出ているエラー名を見て、定義の文字列と一致しているかを確認しておくと安心です。一致していないと、リトライされずにそのまま失敗します。

外側の変数は ItemSelector で渡します。Distributed Mapの子は別の実行として動くので、親で Assign した $modelId を中から直接は参照できません。ItemSelector は親の側で評価されるので、そこで子に渡す入力に詰めています。

256KiBの壁に注意します。Step Functionsでステート間を受け渡せるデータは256KiBまでです。本文が長い記事を入れる場合、JSONLの1行自体は8MBまで読めますが、子に渡す時点で256KiBに収める必要があります。それを超える文書は、本文をS3に置いてキーだけを渡し、Bedrock統合の Input／Output にS3のURIを指定する形に変えてください。

SNSの件名はASCIIで書きます。SNSの Subject は英数字と記号しか使えないので、日本語は本文（Message）側に寄せています。メールで届いたトークンを使って、承認者は次のように返します。

# 承認
aws stepfunctions send-task-success \
  --task-token "メールに記載のトークン" \
  --task-output '{"approved": true}'

# 差し戻し
aws stepfunctions send-task-failure \
  --task-token "メールに記載のトークン" \
  --error Rejected --cause "要約3件に事実誤認あり"

承認者がCLIを使えない場合は、承認・差し戻しボタンのあるページをAPI GatewayとLambdaで用意し、そのLambdaから SendTaskSuccess を呼ぶ形が一般的です。その場合、トークンをURLにそのまま載せるのではなく、DynamoDBに保存して短いIDで引くようにしておくと、メールの転送などで意図しない人に承認される事故を防げます。

いくらかかるか

100記事を要約する場合で試算します。前提は、1記事あたり入力4,000トークン、出力500トークン、子実行1件あたり平均15秒、メモリは課金上の最小単位の64MBとします。Step Functionsの単価はus-east-1の公開価格（2026年10月確認）、LLMの単価は前回のコスト設計の記事と同じClaude Sonnet 5のClaude API価格（入力 $2／出力 $10、100万トークンあたり）を使っています。実測値ではありません。

見てのとおり、費用のほぼすべてはLLMです。Step Functions側はLLMの1%にも届きません。子実行をStandardにしても、1件あたり数回の状態遷移で100件なら1セント以下です。

気をつけたいのはリトライです。Standardではリトライ1回が状態遷移1回として課金され、Expressではリトライの待ち時間もそのまま実行時間として課金されます。金額は小さくても、待ちが積み重なると処理全体が遅くなり、5分の上限にも近づきます。スロットリングを前提に並列数を上げるより、MaxConcurrency を絞って最初から通るようにしたほうが、時間もお金も読みやすくなります。Bedrockの実際の単価はリージョンや推論プロファイルの種類で変わるので、本番の見積もりはBedrockの料金ページで確認してください。

AgentCore統合：エージェントを1ステップとして置く

2026年6月に、Step FunctionsからBedrock AgentCoreのマネージドハーネスを呼ぶ最適化統合 arn:aws:states:::bedrockagentcore:invokeHarness がプレビューで出ました。ハーネスは、モデル、ツール、システムプロンプトを設定として宣言しておくと、ツールを使いながら複数ターンの推論をこなしてくれる実行環境です。Workflow Studioで「AgentCore InvokeHarness」を検索してドラッグすれば、ステートとして置けます。

これで、これまでの「LLMを1回呼ぶステップ」に加えて、「エージェントに調べさせて判断させるステップ」もワークフローの部品になります。たとえば問い合わせの分類までは決まった手順で処理し、分類できなかったものだけエージェントに調べさせ、その結果を人が承認する、という組み方ができます。

ただし、2026年10月時点のドキュメントを読むと、使う前に押さえておくべき制約がかなりあります。

• プレビューで、提供リージョンはバージニア北部、オレゴン、フランクフルト、シドニー。東京リージョンでは使えない
• 統合パターンはRequest Responseのみ。.sync や .waitForTaskToken は使えない
• 1回のステートは最大15分。ステートがタイムアウトしても、ハーネス側は自分のタイムアウトまで動き続ける。実行を止めてもハーネスは止まらない
• 返ってくるのは最後のアシスタント発話のテキストだけ。途中のツール呼び出しや推論の内容は出力に含まれない（CloudWatch側で追う）

3つ目は費用に直結します。ハーネスのタイムアウトを15分より長くしていると、Step Functionsから見ると失敗して終わったのに、裏でエージェントがトークンを使い続けることになります。ドキュメントでも、ハーネスのタイムアウトを15分以内にするよう案内されています。

私の考えでは、手順が決まっている部分まで全部エージェントに任せる必要はありません。要約や分類のように手順が決まっている処理はA〜Cのパターンで組み、判断が要る部分だけをエージェントに渡す。その境目をワークフローの図として残しておけば、あとから「なぜこの結果になったか」を追えます。エージェントの権限設計についてはガードレール設計の記事、Bedrock上のエージェントの基本はBedrock Agents入門に書いています。

運用で見ておくところ

StandardとExpressの使い分け

Expressを使うときに一番詰まりやすいのは、ログを設定し忘れて「失敗したのに何も見えない」状態になることです。子をExpressにしたら、ステートマシンのログ設定でCloudWatch Logsへの出力を有効にしておいてください。

監視する指標

• ExecutionsFailed と ExecutionsTimedOut: 失敗の総数。アラームはまずここに置く
• ExecutionThrottled: Step Functions自体の状態遷移のスロットリング。東京リージョンのStandardは、バケットサイズ・補充レートとも毎秒800（米国の主要リージョンは5,000）
• Distributed Mapの未完了Map Run数とバックログ: 2025年9月から指標が取れるようになった。同時に開けるMap Runは1,000までで、超えた分は待たされる

Bedrock側のスロットリングは、Step Functionsの指標ではなく、リトライの回数として現れます。実行履歴でリトライが目立つようになったら、MaxConcurrency を下げるか、Bedrockのクォータ引き上げを申請するサインです。CloudWatchでのアラーム設計は、CloudWatchのオブザーバビリティ記事の考え方がそのまま使えます。

2万5,000イベントの上限

Standardワークフローの実行履歴は、1実行あたり2万5,000イベントが上限です。超えると実行が失敗します。Distributed Mapではなく通常の（インラインの）Mapで数千件を回したり、Aパターンの修正ループを大きな回数で回したりすると、ここに当たることがあります。件数が多い処理は最初からDistributed Mapにしておくのが無難です。

おわりに

AIの多段処理をワークフローエンジンに載せる理由は、速さではありません。失敗したときに、どこまで終わっていて、何をやり直せばいいかが分かることです。「小さく刻んで並列に回す」は、刻んだ1つ1つを再実行できる形にして初めて運用に耐えるものになります。

始めるなら、ハンズオンの定義から承認ステップを外し、Distributed Mapだけを10件程度のJSONLで動かしてみるのがよいと思います。実行履歴でBedrockの入出力と使用トークン数が見えれば、MaxConcurrency と費用の見当がつきます。承認やエージェントのステップは、その後で足していけば十分です。

※本記事の仕様・料金は2026年10月時点の公開情報に基づいています。AgentCore統合はプレビューのため、仕様や提供リージョンが変わる可能性があります。ASLのサンプルは公式ドキュメントの仕様に沿って作成したもので、筆者の環境でのデプロイ検証は行っていません。費用は仮定した条件による試算です。

参考

• Invoke and customize Amazon Bedrock models with Step Functions
• Invoke Amazon Bedrock AgentCore harness with Step Functions
• AWS Step Functions adds AgentCore-powered agentic reasoning step（2026年6月）
• AWS Step Functions adds 28 new service integrations（2026年3月）
• Durable functions or Step Functions - AWS Lambda
• Handling errors in Step Functions workflows
• Discover service integration patterns in Step Functions
• ItemReader (Map) / ResultWriter (Map)
• Step Functions service quotas
• AWS Step Functions Pricing