📖

Amplify Gen2 + SQS + Lambda + App Runner で LLM API のリトライ機構を実装した話

2026/01/31に公開

はじめに

LLM を使ったチャットアプリを運用していると、「App Runner が再起動したタイミングでユーザーのリクエストが消える」 という問題に直面しました。

本記事では、この問題を Amazon SQS + Lambda + App Runner の構成で解決した実装を紹介します。インフラは Amplify Gen2 (CDK) で定義し、LLM API は FastAPI (Python) で実装しています。

課題

従来のアーキテクチャ

React → App Runner (FastAPI) → Claude API → DynamoDB

問題点

問題	詳細
処理が消える	App Runner の再起動/クラッシュ時、処理中リクエストが消失
リトライ機構がない	一時的な障害でも完全に失敗
障害復旧が困難	失敗リクエストの特定・再処理手段がない

LLM の処理は数十秒〜数分かかるため、その間に App Runner がスケールダウンや再デプロイで再起動すると、ユーザーの質問が回答されないまま消えてしまいます。

解決策

新アーキテクチャ

React
  ↓ POST
API Gateway
  ↓
SQS Queue ←─ 失敗したら自動リトライ（最大3回）
  ↓ トリガー
Lambda (TypeScript)
  ↓ HTTP POST /llm
App Runner (FastAPI)
  ↓
Claude API
  ↓
DynamoDB（結果書き込み）

※ 3回失敗 → DLQ（Dead Letter Queue）→ アラート

なぜこの構成?

評価項目	理由
既存コード活用	App Runner の LLM 処理ロジックをそのまま使用
実装コスト	新規は Lambda（薄いオーケストレーター）のみ
運用コスト	SQS + Lambda はサーバーレスで低コスト
障害耐性	SQS 自動リトライ + DLQ で失敗リクエスト保持

実装

ディレクトリ構成

amplify/
├── backend.ts                    # SQS/Lambda/API Gateway 定義
└── functions/
    └── llm-worker/
        ├── resource.ts           # Lambda 定義
        ├── handler.ts            # Lambda ハンドラー
        └── package.json

# App Runner (別リポジトリ)
app-runner-llm-api/
└── src/
    └── app.py                    # /llm エンドポイント

1. 依存関係の追加

Amplify Gen2 で SQS + Lambda + API Gateway を使うには、CDK 関連パッケージが必要です。

# pnpm
pnpm add -D aws-cdk-lib constructs @types/aws-lambda

# npm
npm install --save-dev aws-cdk-lib constructs @types/aws-lambda

2. Lambda 関数の定義

// amplify/functions/llm-worker/resource.ts
import { defineFunction } from '@aws-amplify/backend';

export const llmWorker = defineFunction({
  name: 'llm-worker',
  entry: './handler.ts',
  timeoutSeconds: 300, // 5分（LLM処理時間考慮）
  memoryMB: 512,
});

// amplify/functions/llm-worker/package.json
{
  "name": "llm-worker",
  "type": "module",
  "dependencies": {
    "@aws-sdk/client-dynamodb": "^3.700.0",
    "@aws-sdk/lib-dynamodb": "^3.700.0"
  },
  "devDependencies": {
    "@types/aws-lambda": "^8.10.145"
  }
}

3. backend.ts（SQS + Lambda + API Gateway）

// amplify/backend.ts
import { defineBackend } from '@aws-amplify/backend';
import { auth } from './auth/resource';
import { data } from './data/resource';
import { llmWorker } from './functions/llm-worker/resource';
import { PolicyStatement, Role, ServicePrincipal } from 'aws-cdk-lib/aws-iam';
import { Function } from 'aws-cdk-lib/aws-lambda';
import * as sqs from 'aws-cdk-lib/aws-sqs';
import * as cdk from 'aws-cdk-lib';
import * as lambdaEventSources from 'aws-cdk-lib/aws-lambda-event-sources';
import * as apigateway from 'aws-cdk-lib/aws-apigateway';

const backend = defineBackend({
  auth,
  data,
  llmWorker,
});

// ========================================
// SQS + Lambda 構成
// ========================================

const sqsStack = backend.createStack('LlmJobSqs');

// DLQ（Dead Letter Queue）
const llmJobDlq = new sqs.Queue(sqsStack, 'LlmJobDlq', {
  retentionPeriod: cdk.Duration.days(14),
});

// メインキュー
const llmJobQueue = new sqs.Queue(sqsStack, 'LlmJobQueue', {
  visibilityTimeout: cdk.Duration.minutes(6), // Lambda タイムアウト + バッファ
  retentionPeriod: cdk.Duration.days(7),
  deadLetterQueue: {
    queue: llmJobDlq,
    maxReceiveCount: 3, // 3回失敗で DLQ へ
  },
});

// Lambda に SQS トリガーを追加
const llmWorkerLambda = backend.llmWorker.resources.lambda as Function;
llmWorkerLambda.addEventSource(
  new lambdaEventSources.SqsEventSource(llmJobQueue, {
    batchSize: 1,
  }),
);

// 環境変数
llmWorkerLambda.addEnvironment(
  'APP_RUNNER_LLM_ENDPOINT',
  process.env.APP_RUNNER_LLM_ENDPOINT || '',
);

// DynamoDB アクセス権限
const chatMessageTable = backend.data.resources.tables['ChatMessage'];
const llmJobTable = backend.data.resources.tables['LlmJob'];

llmWorkerLambda.addEnvironment('CHAT_MESSAGE_TABLE_NAME', chatMessageTable.tableName);
llmWorkerLambda.addEnvironment('LLM_JOB_TABLE_NAME', llmJobTable.tableName);

llmWorkerLambda.addToRolePolicy(
  new PolicyStatement({
    actions: ['dynamodb:GetItem', 'dynamodb:PutItem', 'dynamodb:UpdateItem'],
    resources: [chatMessageTable.tableArn, llmJobTable.tableArn],
  }),
);

// ========================================
// API Gateway → SQS 連携
// ========================================

const apiGatewayRole = new Role(sqsStack, 'ApiGatewaySqsRole', {
  assumedBy: new ServicePrincipal('apigateway.amazonaws.com'),
});
llmJobQueue.grantSendMessages(apiGatewayRole);

const api = new apigateway.RestApi(sqsStack, 'LlmJobApi', {
  restApiName: 'LLM Job API',
  defaultCorsPreflightOptions: {
    allowOrigins: apigateway.Cors.ALL_ORIGINS,
    allowMethods: ['POST', 'OPTIONS'],
    allowHeaders: ['Content-Type', 'Authorization'],
  },
});

// SQS 統合
const sqsIntegration = new apigateway.AwsIntegration({
  service: 'sqs',
  path: `${cdk.Aws.ACCOUNT_ID}/${llmJobQueue.queueName}`,
  integrationHttpMethod: 'POST',
  options: {
    credentialsRole: apiGatewayRole,
    requestParameters: {
      'integration.request.header.Content-Type': "'application/x-www-form-urlencoded'",
    },
    requestTemplates: {
      'application/json': 'Action=SendMessage&MessageBody=$input.body',
    },
    integrationResponses: [
      {
        statusCode: '200',
        responseParameters: {
          'method.response.header.Access-Control-Allow-Origin': "'*'",
        },
        responseTemplates: {
          'application/json':
            '{"status": "accepted", "messageId": "$input.path(\'$.SendMessageResponse.SendMessageResult.MessageId\')"}',
        },
      },
    ],
  },
});

// POST /jobs エンドポイント
const jobsResource = api.root.addResource('jobs');
jobsResource.addMethod('POST', sqsIntegration, {
  methodResponses: [
    {
      statusCode: '200',
      responseParameters: {
        'method.response.header.Access-Control-Allow-Origin': true,
      },
    },
  ],
});

// Outputs
new cdk.CfnOutput(sqsStack, 'LlmJobApiEndpoint', {
  value: api.url,
  description: 'API Gateway endpoint',
});

4. Lambda ハンドラー

// amplify/functions/llm-worker/handler.ts
import type { SQSEvent } from 'aws-lambda';
import { DynamoDBClient } from '@aws-sdk/client-dynamodb';
import { DynamoDBDocumentClient, UpdateCommand } from '@aws-sdk/lib-dynamodb';

const APP_RUNNER_LLM_ENDPOINT = process.env.APP_RUNNER_LLM_ENDPOINT || '';
const CHAT_MESSAGE_TABLE_NAME = process.env.CHAT_MESSAGE_TABLE_NAME || '';
const LLM_JOB_TABLE_NAME = process.env.LLM_JOB_TABLE_NAME || '';

const dynamoClient = new DynamoDBClient({});
const docClient = DynamoDBDocumentClient.from(dynamoClient);

interface LlmJobMessage {
  sessionId: string;
  jobId: string;
  messageId: string;
  userMessage: string;
}

interface LlmApiResponse {
  content: string;
  model?: string;
  usage?: { input_tokens: number; output_tokens: number };
}

// App Runner の /llm を呼び出す
async function callLlmApi(userMessage: string): Promise<LlmApiResponse> {
  const response = await fetch(`${APP_RUNNER_LLM_ENDPOINT}/llm`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ userMessage }),
  });

  if (!response.ok) {
    throw new Error(`LLM API error: ${response.status}`);
  }

  return await response.json();
}

// DynamoDB 更新ヘルパー
async function updateLlmJobStatus(jobId: string, status: string, resultPayload?: object) {
  const updateExpression = ['#status = :status', 'updatedAt = :updatedAt'];
  const values: Record<string, unknown> = {
    ':status': status,
    ':updatedAt': new Date().toISOString(),
  };

  if (resultPayload) {
    updateExpression.push('resultPayload = :resultPayload');
    values[':resultPayload'] = JSON.stringify(resultPayload);
  }

  await docClient.send(
    new UpdateCommand({
      TableName: LLM_JOB_TABLE_NAME,
      Key: { id: jobId },
      UpdateExpression: `SET ${updateExpression.join(', ')}`,
      ExpressionAttributeNames: { '#status': 'status' },
      ExpressionAttributeValues: values,
    }),
  );
}

async function updateChatMessage(messageId: string, status: string, content?: string) {
  const updateExpression = ['#status = :status', 'updatedAt = :updatedAt'];
  const values: Record<string, unknown> = {
    ':status': status,
    ':updatedAt': new Date().toISOString(),
  };

  if (content) {
    updateExpression.push('content = :content');
    values[':content'] = content;
  }

  await docClient.send(
    new UpdateCommand({
      TableName: CHAT_MESSAGE_TABLE_NAME,
      Key: { id: messageId },
      UpdateExpression: `SET ${updateExpression.join(', ')}`,
      ExpressionAttributeNames: { '#status': 'status' },
      ExpressionAttributeValues: values,
    }),
  );
}

// Lambda ハンドラー
export const handler = async (event: SQSEvent): Promise<void> => {
  for (const record of event.Records) {
    const body: LlmJobMessage = JSON.parse(record.body);
    const { jobId, messageId, userMessage } = body;

    console.log(`Processing job: ${jobId}`);

    try {
      await updateLlmJobStatus(jobId, 'running');

      const llmResponse = await callLlmApi(userMessage);

      await updateChatMessage(messageId, 'completed', llmResponse.content);
      await updateLlmJobStatus(jobId, 'succeeded', {
        answer: llmResponse.content,
        model: llmResponse.model,
        usage: llmResponse.usage,
      });

      console.log(`Job completed: ${jobId}`);
    } catch (error) {
      console.error(`Job failed: ${jobId}`, error);

      await updateChatMessage(messageId, 'error', 'エラーが発生しました');
      await updateLlmJobStatus(jobId, 'failed');

      // 例外を投げて SQS にリトライさせる
      throw error;
    }
  }
};

5. App Runner (FastAPI) の /llm エンドポイント

# src/app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class LlmRequest(BaseModel):
    userMessage: str

class LlmResponse(BaseModel):
    content: str
    model: str | None = None
    usage: dict | None = None

@app.post("/llm", response_model=LlmResponse)
async def handle_llm(request: LlmRequest):
    """
    LLM 処理専用エンドポイント
    Lambda がこのエンドポイントを呼び出す。
    DB 書き込みは Lambda 側で行う。
    """
    try:
        llm_service = get_llm_service()  # 既存のLLMサービス
        response = await llm_service.generate(request.userMessage)

        return LlmResponse(
            content=response.content,
            model=response.model,
            usage=response.usage,
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

6. amplify.yml（ビルド設定）

version: 1
backend:
  phases:
    build:
      commands:
        - npm install
        # Lambda 関数の依存関係を個別にインストール
        - cd amplify/functions/llm-worker && npm install && cd ../../..
        - npx ampx pipeline-deploy --branch $AWS_BRANCH --app-id $AWS_APP_ID
frontend:
  phases:
    build:
      commands:
        - npm run build
  artifacts:
    baseDirectory: dist
    files:
      - '**/*'

7. フロントエンド（React）

// API Gateway → SQS に送信
export const submitLlmJob = async (request: LlmJobRequest) => {
  const endpoint = import.meta.env.VITE_LLM_API_GATEWAY_ENDPOINT;

  const response = await fetch(`${endpoint}jobs`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      sessionId: request.sessionId,
      jobId: request.jobId,
      messageId: request.messageId,
      userMessage: request.prompt,
    }),
  });

  if (!response.ok) {
    throw new Error(`API Gateway error: ${response.status}`);
  }

  // 即座に 202 Accepted が返る
  // 実際の結果は DynamoDB をポーリングして取得
  return await response.json();
};

ハマりポイント・教訓

1. SQS キュー名の 80 文字制限

問題: queueName を明示的に指定すると、Amplify が環境名を付加して 80 文字を超えることがある。

解決: queueName は指定せず、CDK に自動生成させる。

// ❌ NG
const queue = new sqs.Queue(stack, 'LlmJobQueue', {
  queueName: 'llm-job-queue', // 80文字超過の可能性
});

// ✅ OK
const queue = new sqs.Queue(stack, 'LlmJobQueue', {
  // queueName は指定しない
});

2. Lambda 依存関係のインストール

問題: Amplify ビルド時に Lambda 関数の node_modules がインストールされない。

解決: amplify.yml で明示的に npm install を実行。

- cd amplify/functions/llm-worker && npm install && cd ../../..

3. Visibility Timeout の設定

問題: Lambda タイムアウト（5分）より Visibility Timeout が短いと、処理中にメッセージが再度 visible になりリトライが発生する。

解決: Visibility Timeout = Lambda タイムアウト + バッファ（1分）

const llmJobQueue = new sqs.Queue(sqsStack, 'LlmJobQueue', {
  visibilityTimeout: cdk.Duration.minutes(6), // Lambda 5分 + バッファ 1分
});

4. API Gateway エンドポイントの取得

デプロイ後、CloudFormation Outputs から取得:

aws cloudformation describe-stacks \
  --stack-name "amplify-xxx-LlmJobSqs-xxx" \
  --query "Stacks[0].Outputs[?OutputKey=='LlmJobApiEndpoint'].OutputValue" \
  --output text

リトライの動作

正常フロー

React → API Gateway → SQS → Lambda → App Runner /llm → DynamoDB
                              ↓
                        (成功) メッセージ削除

リトライフロー

SQS → Lambda → App Runner /llm
        ↓
      (失敗) 例外スロー
        ↓
SQS (Visibility Timeout 経過後)
        ↓
      リトライ 1回目 → 成功 → メッセージ削除

DLQ フロー

SQS → Lambda (失敗 1回目)
  ↓
リトライ 1 (失敗 2回目)
  ↓
リトライ 2 (失敗 3回目)
  ↓
DLQ へ移動 → CloudWatch アラート

まとめ

SQS + Lambda を挟むことで、以下のメリットが得られました：

Before	After
App Runner 障害で処理消失	SQS がメッセージを保持、自動リトライ
リトライ手段なし	最大 3 回の自動リトライ
失敗リクエストが不明	DLQ で失敗リクエストを保持

Amplify Gen2 の CDK オーバーライド機能を使えば、SQS や API Gateway も同じリポジトリで管理できるので、インフラとアプリケーションの一貫性が保てます。