AIP-C01 06: 本番統合とレジリエンス

Mix.install([
  {:aws, "~> 1.0.15"},
  {:hackney, "~> 4.7"},
  {:kino, "~> 0.19"},
  {:kino_vega_lite, "~> 0.1"},
  {:jason, "~> 1.4"}
])

概要

最終確認日: 2026-09-22

対応範囲: 1.1, 1.2, 2.2〜2.5, 4.2, 5.2

このノートで身につけること

• 同期、ストリーミング、非同期、バッチを要件から選ぶ
• timeout、retry、backoff、jitter、circuit breaker を設計する
• API Gateway、Lambda、SQS、EventBridge、Step Functions を疎結合に組む
• カナリア、品質ゲート、ロールバックを含む GenAI CI/CD を考える

最新仕様の要点

Bedrock の batch inference は大量の独立プロンプトを S3 入出力で非同期処理します。Converse 形式も利用できますが、tool calling、structured output、複数ターンのやり取りは使えません。状態変更は EventBridge で受け、ポーリングを避けられます。

スロットリングや一時的容量不足では、Retry-After を尊重し、指数バックオフと jitter、上限付き試行回数、同時実行制御を使います。429 をすべて同じ原因と決めつけず、エラー型、RPM / TPM、max_tokens、モデル準備状態を確認します。503 / 529 は一時的容量エラーとして扱い、必要なら cross-Region inference やフォールバックを検討します。

AWS 接続と実習タグ

後半の SQS 実習で2つのキューを作成します。長期アクセスキーを保存せず、一時認証情報または Livebook Secrets を使ってください。

access_key_input = Kino.Input.password("AWS_ACCESS_KEY_ID")
secret_key_input = Kino.Input.password("AWS_SECRET_ACCESS_KEY")
session_token_input = Kino.Input.password("AWS_SESSION_TOKEN（一時認証の場合）")
region_input = Kino.Input.text("AWS_REGION", default: "ap-northeast-1")
owner_input = Kino.Input.text("aip:owner（英数字・ハイフン）", default: "student")

Kino.Layout.grid(
  [access_key_input, secret_key_input, session_token_input, region_input, owner_input],
  columns: 2
)

access_key = Kino.Input.read(access_key_input)
secret_key = Kino.Input.read(secret_key_input)
session_token = Kino.Input.read(session_token_input)
region = Kino.Input.read(region_input)

client =
  if session_token in [nil, ""] do
    AWS.Client.create(access_key, secret_key, region)
  else
    AWS.Client.create(access_key, secret_key, session_token, region)
  end

owner =
  owner_input
  |> Kino.Input.read()
  |> String.replace(~r/[^A-Za-z0-9_-]/, "-")
  |> String.slice(0, 20)

lab_id = "#{Date.utc_today() |> Date.to_iso8601() |> String.replace("-", "")}-#{System.unique_integer([:positive])}"
expires_at = DateTime.utc_now() |> DateTime.add(4, :hour) |> DateTime.to_iso8601()

queue_tags = %{
  "aip:managed-by" => "livebook",
  "aip:lab-id" => lab_id,
  "aip:notebook" => "06-production-integration",
  "aip:owner" => owner,
  "aip:expires-at" => expires_at
}

{:ok, queue_ledger} = Agent.start_link(fn -> [] end)

%{lab_id: lab_id, expires_at: expires_at, tags: queue_tags}

ハンズオン1: 統合方式を要件から選ぶ

select_pattern = fn request ->
  cond do
    request.count > 1_000 and request.deadline_minutes >= 60 -> :batch_inference
    request.user_facing and request.first_token_important -> :streaming
    request.long_running -> :async_job_with_eventbridge
    request.needs_tools or request.multi_turn -> :converse
    true -> :synchronous_invoke
  end
end

workloads = [
  %{name: "チャット", count: 1, deadline_minutes: 1, user_facing: true, first_token_important: true, long_running: false, needs_tools: false, multi_turn: true},
  %{name: "夜間100万件分類", count: 1_000_000, deadline_minutes: 480, user_facing: false, first_token_important: false, long_running: true, needs_tools: false, multi_turn: false},
  %{name: "社内ツール実行", count: 1, deadline_minutes: 2, user_facing: true, first_token_important: false, long_running: false, needs_tools: true, multi_turn: true}
]

workloads
|> Enum.map(&Map.put(&1, :pattern, select_pattern.(&1)))
|> Kino.DataTable.new()

ハンズオン2: 指数バックオフと full jitter

defmodule RetryPolicy do
  @transient [:throttling, :service_unavailable, :model_not_ready, :overloaded]

  def run(fun, opts \\ []) do
    max_attempts = Keyword.get(opts, :max_attempts, 6)
    base_ms = Keyword.get(opts, :base_ms, 100)
    cap_ms = Keyword.get(opts, :cap_ms, 2_000)
    do_run(fun, 1, max_attempts, base_ms, cap_ms)
  end

  defp do_run(fun, attempt, max_attempts, base_ms, cap_ms) do
    case fun.() do
      {:ok, value} -> {:ok, value, attempt}
      {:error, reason} when reason in @transient and attempt < max_attempts ->
        max_delay = min(cap_ms, trunc(base_ms * :math.pow(2, attempt - 1)))
        delay = :rand.uniform(max_delay)
        Process.sleep(delay)
        do_run(fun, attempt + 1, max_attempts, base_ms, cap_ms)

      {:error, reason} ->
        {:error, reason, attempt}
    end
  end
end

{:ok, counter} = Agent.start_link(fn -> 0 end)

unstable_call = fn ->
  attempt = Agent.get_and_update(counter, fn value -> {value + 1, value + 1} end)
  if attempt < 3, do: {:error, :throttling}, else: {:ok, "success"}
end

RetryPolicy.run(unstable_call, base_ms: 10, cap_ms: 100)

実運用では AWS SDK の標準 retry を優先し、アプリ層の retry と二重に増幅しないようにします。認証エラー、入力不正、AccessDenied のような恒久エラーは再試行しません。

backoff_rows =
  Enum.map(1..6, fn attempt ->
    %{
      attempt: attempt,
      max_delay_ms: min(2_000, trunc(100 * :math.pow(2, attempt - 1)))
    }
  end)

VegaLite.new(width: 620, height: 250, title: "指数バックオフの上限（実待機は 0〜上限の jitter）")
|> VegaLite.data_from_values(backoff_rows)
|> VegaLite.mark(:line, point: true, tooltip: true)
|> VegaLite.encode_field(:x, "attempt", type: :ordinal, title: "試行回数")
|> VegaLite.encode_field(:y, "max_delay_ms", type: :quantitative, title: "最大待機 (ms)")
|> Kino.VegaLite.new()

ハンズオン3: circuit breaker の状態遷移

transition = fn
  %{state: :closed, failures: failures} = breaker, :failure when failures + 1 >= 3 ->
    %{breaker | state: :open, failures: failures + 1}

  %{state: :closed, failures: failures} = breaker, :failure ->
    %{breaker | failures: failures + 1}

  %{state: :closed} = breaker, :success ->
    %{breaker | failures: 0}

  %{state: :open} = breaker, :cooldown_elapsed ->
    %{breaker | state: :half_open}

  %{state: :half_open} = breaker, :success ->
    %{breaker | state: :closed, failures: 0}

  %{state: :half_open} = breaker, :failure ->
    %{breaker | state: :open}

  breaker, _event ->
    breaker
end

events = [:failure, :failure, :failure, :cooldown_elapsed, :success]

{_last, history} =
  Enum.map_reduce(events, %{state: :closed, failures: 0}, fn event, breaker ->
    next = transition.(breaker, event)
    {%{event: event, before: breaker.state, after: next.state}, next}
  end)

history

フォールバックは「別モデルへ黙って切り替える」だけではありません。キャッシュ済み応答、検索結果だけ、受付だけして後処理、人間窓口への誘導など、品質と業務リスクに合わせます。

ハンズオン4: EventBridge の batch 完了イベントを分類する

handle_event = fn event ->
  case {event["source"], event["detail-type"], get_in(event, ["detail", "status"])} do
    {"aws.bedrock", "Batch Inference Job State Change", "Completed"} ->
      {:process_output, get_in(event, ["detail", "jobArn"])}

    {"aws.bedrock", "Batch Inference Job State Change", status}
    when status in ["Failed", "Stopped"] ->
      {:alert, status}

    _ ->
      :ignore
  end
end

sample_event = %{
  "source" => "aws.bedrock",
  "detail-type" => "Batch Inference Job State Change",
  "detail" => %{"status" => "Completed", "jobArn" => "arn:aws:bedrock:REGION:ACCOUNT:model-invocation-job/ID"}
}

handle_event.(sample_event)

ハンズオン5: DLQ 付き非同期キューを構築する

ユーザーへの同期応答と長時間処理を分離する最小構成を SQS で作ります。DLQ を先に作り、その ARN を source queue の RedrivePolicy に設定します。タグは作成リクエストに含め、タグなしの瞬間を作りません。

queue_prefix = "aip-#{owner}-#{lab_id}"
dlq_name = "#{queue_prefix}-dlq"
work_queue_name = "#{queue_prefix}-work"

{:ok, %{"QueueUrl" => dlq_url}, _} =
  AWS.SQS.create_queue(client, %{
    "QueueName" => dlq_name,
    "Attributes" => %{"MessageRetentionPeriod" => "3600"},
    "tags" => queue_tags
  })

Agent.update(queue_ledger, fn resources ->
  [%{type: :queue, role: :dlq, name: dlq_name, url: dlq_url} | resources]
end)

Process.sleep(1_000)

{:ok, %{"Attributes" => %{"QueueArn" => dlq_arn}}, _} =
  AWS.SQS.get_queue_attributes(client, %{
    "QueueUrl" => dlq_url,
    "AttributeNames" => ["QueueArn"]
  })

{:ok, %{"QueueUrl" => work_queue_url}, _} =
  AWS.SQS.create_queue(client, %{
    "QueueName" => work_queue_name,
    "Attributes" => %{
      "VisibilityTimeout" => "10",
      "MessageRetentionPeriod" => "3600",
      "ReceiveMessageWaitTimeSeconds" => "2",
      "RedrivePolicy" => Jason.encode!(%{
        "deadLetterTargetArn" => dlq_arn,
        "maxReceiveCount" => "3"
      })
    },
    "tags" => queue_tags
  })

Agent.update(queue_ledger, fn resources ->
  [%{type: :queue, role: :work, name: work_queue_name, url: work_queue_url} | resources]
end)

Process.sleep(1_000)

Kino.DataTable.new(Agent.get(queue_ledger, &Enum.reverse/1))

作成セルの途中で失敗した場合も、成功したキューは直後に台帳へ入ります。IAM には sqs:CreateQueue に加えて、作成時タグ用の sqs:TagQueue が必要です。

job = %{
  job_id: "job-001",
  operation: "summarize",
  input_uri: "s3://example-input/prefix/item.jsonl",
  idempotency_key: "tenant-a:item-42:v1",
  trace_id: "trace-001"
}

{:ok, sent, _} =
  AWS.SQS.send_message(client, %{
    "QueueUrl" => work_queue_url,
    "MessageBody" => Jason.encode!(job)
  })

{:ok, received, _} =
  AWS.SQS.receive_message(client, %{
    "QueueUrl" => work_queue_url,
    "MaxNumberOfMessages" => 1,
    "WaitTimeSeconds" => 2,
    "AttributeNames" => ["ApproximateReceiveCount"]
  })

%{sent: sent, received: received}

本番 worker は処理が成功した後だけ DeleteMessage を呼びます。可視性タイムアウトより処理が長い場合は延長し、重複配信を前提に idempotency key を永続化します。DLQ へ到達したメッセージは「自動的に直った」ことにはならないため、アラーム、原因分類、redrive 前の修正が必要です。

{:ok, %{"Attributes" => queue_attributes}, _} =
  AWS.SQS.get_queue_attributes(client, %{
    "QueueUrl" => work_queue_url,
    "AttributeNames" => [
      "QueueArn",
      "VisibilityTimeout",
      "RedrivePolicy",
      "ApproximateNumberOfMessages",
      "ApproximateNumberOfMessagesNotVisible"
    ]
  })

{:ok, %{"Tags" => actual_queue_tags}, _} =
  AWS.SQS.list_queue_tags(client, %{"QueueUrl" => work_queue_url})

Kino.Layout.tabs([
  {"キュー属性", Kino.DataTable.new([queue_attributes])},
  {"タグ", Kino.DataTable.new(Enum.map(actual_queue_tags, fn {key, value} -> %{key: key, value: value} end))}
])

リファレンスアーキテクチャ

Kino.Mermaid.new("""
flowchart
  C["Client"] --> W["WAF / Cognito"]
  W --> A["API Gateway"]
  A --> L["Lambda / ECS"]
  L --> G["Guardrail input check"]
  G --> B["Bedrock Converse / Stream"]
  L --> D["DynamoDB conversation state"]
  L --> Q["SQS overflow / async"]
  L --> S["Step Functions tools / approval"]

  L -.-> O["CloudWatch / X-Ray / CloudTrail"]
  B -.-> O
  S -.-> O

  SI["S3 batch input"] --> J["Model invocation job"]
  J --> E["EventBridge"]
  E --> P["Output processor"]
  P --> SO["S3 batch output"]
""")

CI/CD 品質ゲート

1. IaC と IAM policy の静的検査
2. prompt / schema / tool contract のユニットテスト
3. 固定評価セットで品質・安全性・検索・コスト・レイテンシー評価
4. staging で synthetic workflow
5. canary で少量トラフィック
6. SLO と品質閾値を満たさなければ自動ロールバック

判断問題

1. 429 時に無制限 retry が障害を悪化させる仕組みを説明してください。
2. 大量の独立要約に batch inference が向き、tool use に向かない理由は何ですか。
3. API Gateway のタイムアウトより FM の最悪レイテンシーが長い場合、どんな統合へ変更しますか。
4. model fallback で品質・安全・データ所在地を再評価すべき理由は何ですか。

公式資料

• Scaling and throughput best practices
• Batch inference
• Bedrock events with EventBridge
• Cross-Region inference
• Latency-optimized inference
• SQS CreateQueue
• SQS dead-letter queues
• Resource Groups Tagging API: GetResources

クリーンアップ: 必ず実行する

source queue が DLQ を参照しているため、source → DLQ の順に削除します。SQS はキュー削除後、同名キューを再作成できるまで最低60秒必要です。

created_queues = Agent.get(queue_ledger, &Enum.reverse/1)
Kino.DataTable.new(created_queues)

cleanup_results =
  created_queues
  |> Enum.reverse()
  |> Enum.map(fn queue ->
    %{name: queue.name, role: queue.role, result: AWS.SQS.delete_queue(client, %{"QueueUrl" => queue.url})}
  end)

Kino.DataTable.new(
  Enum.map(cleanup_results, fn row -> Map.update!(row, :result, &inspect/1) end)
)

service_audit =
  Enum.reduce_while(1..30, [], fn _attempt, _last ->
    case AWS.SQS.list_queues(client, %{"QueueNamePrefix" => queue_prefix}) do
      {:ok, body, _} ->
        remaining = Map.get(body, "QueueUrls", [])

        if remaining == [] do
          {:halt, []}
        else
          Process.sleep(1_000)
          {:cont, remaining}
        end

      error ->
        {:halt, [%{"auditError" => inspect(error)}]}
    end
  end)

tag_audit =
  case AWS.ResourceGroupsTaggingAPI.get_resources(client, %{
         "TagFilters" => [
           %{"Key" => "aip:managed-by", "Values" => ["livebook"]},
           %{"Key" => "aip:lab-id", "Values" => [lab_id]}
         ],
         "ResourceTypeFilters" => ["sqs:queue"]
       }) do
    {:ok, body, _} -> Map.get(body, "ResourceTagMappingList", [])
    error -> [%{"auditError" => inspect(error)}]
  end

summary = %{
  lab_id: lab_id,
  service_resources_remaining: length(service_audit),
  tagged_resources_remaining: length(tag_audit),
  cleanup_complete: service_audit == [] and tag_audit == []
}

Kino.Layout.tabs([
  {"結果", Kino.DataTable.new([summary])},
  {"SQS 残存", Kino.DataTable.new(Enum.map(service_audit, &%{queue_url: &1}))},
  {"タグ残存", Kino.DataTable.new(tag_audit)}
])

cleanup_complete が true になるまで実習は完了ではありません。タグ検索には反映遅延があるため、サービス固有の ListQueues が空でタグ側だけ残る場合は、少し待って監査セルを再実行してください。台帳を失った場合は aip:managed-by=livebook と aip:owner=<自分> で期限切れリソースを検索します。