🐾

Amazon Bedrockでプロダクション品質のマルチエージェントシステムを実現する

に公開

はじめに

header

課題提起

Amazon Bedrock を活用したマルチエージェント型の LLM システム開発において、以下のような課題に直面していないでしょうか?

  • LLM出力の品質保証をどう担保するか - 非決定的な出力をどうテストすべきか
  • Rate Limit制約下での設計 - Claude Sonnet 4.5の2 RPM制約にどう対処するか
  • トレーサビリティの確保 - 生成コンテンツの信頼性をどう証明するか
  • エージェント間の責務分離 - 複雑化するワークフローをどう整理するか

記事の目的

本記事では、マルチエージェント型のコンテンツ生成システムを開発した際に得た、以下の知見を共有します。

  1. Fact-First Architecture によるトレーサビリティ設計
  2. Property-Based Testing による LLM 出力の品質保証
  3. Claude 4.5 シリーズの使い分けパターン(Sonnet/Opus/Haiku)
  4. Rate Limit 制約をアーキテクチャで解決する設計思想

この記事を読むことで、プロダクション環境で運用可能な Bedrock ベースのマルチエージェントシステムの設計指針を理解できます。


背景と全体像

システム概要

本記事で扱うのは、情報収集から記事生成、品質評価、公開までを完全自動化するマルチエージェント型システムです。

アーキテクチャの特徴

  • インフラ: AWS Serverless(Lambda Python 3.12, Step Functions, DynamoDB, S3)
  • AI モデル: Amazon Bedrock(Claude 4.5 シリーズ)
    • Claude Sonnet 4.5 - 記事生成・オーケストレーション
    • Claude Opus 4.5 - 高難度タスク(長大コンテキスト要約)
    • Claude Haiku 4.5 - 校閲(高速・低コスト)
  • IaC: AWS SAM
  • テスト: pytest + Hypothesis(Property-Based Testing)

エージェント思想による 8 層構成

本システムは、Amazon Bedrock Agents(マネージドサービス)そのものは使用していませんが、エージェント設計思想に基づいて構築しています。各 Lambda 関数が特定の責務を持つ「エージェント」として機能し、Step Functions がオーケストレーションを担当します。

この多層構造により、各エージェント(Lambda 関数)が明確な責務を持ち、保守性の高い設計を実現しています。

Step Functions


詳細な解説・分析

1. Fact-First Architecture - トレーサビリティ重視の設計

1.1 なぜ Fact-First なのか

従来の LLM コンテンツ生成では、「情報源が不明確」「事実とフィクションの混在」が課題でした。
本システムでは、すべての記述を検証可能な事実に基づかせる設計を採用しました。

1.2 Fact Layer の 3 段階処理

Fact Layer は、以下の 3 段階で事実を抽出・統合・要約します。

F1: Fact Extraction(事実抽出エージェント)

# 出力例
{
  "claim": "Claude Sonnet 4.5のリージョン別利用可能性が2025年1月に拡大した",
  "confidence": 0.95,
  "source_url": "https://aws.amazon.com/bedrock/...",
  "snippet": "Claude Sonnet 4.5 is now available in...",
  "timestamp": "2025-01-15T12:00:00Z"
}

ポイント:

  • 単一の検証可能な主張(claim)に分解
  • 出典 URL とスニペットを必須化
  • 信頼度スコアを付与

F2: Fact Normalization & Bundling(事実正規化エージェント)

# 複数ソースから同一事実を統合
{
  "canonical_claim": "Claude Sonnet 4.5が2025年1月にAPAC地域で利用可能になった",
  "source_urls": [
    "https://aws.amazon.com/...",
    "https://docs.anthropic.com/..."
  ],
  "aggregated_confidence": 0.97
}

ポイント:

  • 類似する主張を正規化して統合
  • 複数ソースによる相互検証
  • 矛盾する主張は別ファクトとして分離

F3: Fact Summarization(事実要約エージェント)

{
  "summary": "Claude Sonnet 4.5のAPAC展開により...",
  "key_numbers": ["2 RPM", "200K tokens"],
  "key_entities": ["Amazon Bedrock", "Claude 4.5", "APAC"],
  "contentious_points": []
}

効果:

  • 記事生成時にファクトのみを参照することで、ハルシネーションを防止
  • 出典への完全なトレーサビリティを確保

1.3 実装パターン: Pydantic によるデータモデル定義

from pydantic import BaseModel, Field, HttpUrl
from decimal import Decimal

class Fact(BaseModel):
    """検証可能な単一の事実"""
    claim: str = Field(..., min_length=10)
    confidence: Decimal = Field(..., ge=0.0, le=1.0)
    source_url: HttpUrl
    snippet: str
    timestamp: str

    class Config:
        # Decimal型をDynamoDBで扱う
        json_encoders = {Decimal: float}

2. Property-Based Testing - LLM 出力の品質保証

2.1 従来のテスト手法の限界

LLM の出力は非決定的であり、従来の「期待値との完全一致」テストは機能しません。
そこで、満たすべき性質(Property)をテストするアプローチを採用しました。

2.2 Hypothesis によるプロパティテスト

テストケース 1: Fact モデルの不変条件

from hypothesis import given
from hypothesis import strategies as st

@given(
    claim=st.text(min_size=10),
    confidence=st.decimals(min_value=0.0, max_value=1.0, places=2)
)
def test_fact_invariants(claim: str, confidence: Decimal):
    """Factモデルが常に満たすべき性質"""
    fact = Fact(
        claim=claim,
        confidence=confidence,
        source_url="https://example.com",
        snippet="test snippet",
        timestamp="2025-01-01T00:00:00Z"
    )

    # 不変条件1: 信頼度は0-1の範囲
    assert 0.0 <= fact.confidence <= 1.0

    # 不変条件2: claimは最小長を満たす
    assert len(fact.claim) >= 10

テストケース 2: FactBundle の集約規則

@given(st.lists(st.from_type(Fact), min_size=2, max_size=10))
def test_fact_bundle_aggregation(facts: list[Fact]):
    """複数のFactを統合する際の性質"""
    bundle = FactBundle.from_facts(facts)

    # 性質1: すべてのソースURLが保持される
    assert len(bundle.source_urls) == len(set(f.source_url for f in facts))

    # 性質2: 信頼度は個別Factの最大値以下
    assert bundle.aggregated_confidence <= max(f.confidence for f in facts)

3. Claude 4.5 シリーズの使い分けパターン

3.1 モデル選択の基本方針

Claude 4.5 シリーズは、用途に応じて以下のように使い分けます。

モデル 用途 理由
Sonnet 4.5 Fact抽出、記事生成、オーケストレーション バランス型(精度・速度・コスト)
Opus 4.5 長大コンテキスト要約、複雑な構造設計 最高精度(200Kトークン対応)
Haiku 4.5 校閲、簡易チェック 高速・低コスト

Bedrock Model

3.2 実装パターン: モデル選択の動的切り替え

from functools import lru_cache

AGENT_MODEL_PREFERENCES = {
    "fact_extraction": ("SONNET", "OPUS"),  # Fact抽出: Sonnet優先、失敗時Opus
    "article_writer": ("SONNET",),          # 記事生成: Sonnetのみ
    "proofreader": ("HAIKU", "SONNET"),     # 校閲: Haiku優先、複雑ケースでSonnet
}

@lru_cache(maxsize=128)
def get_model_for_agent(agent_id: str) -> str:
    """エージェントIDに基づいてモデルを選択"""
    preferences = AGENT_MODEL_PREFERENCES.get(agent_id, ("SONNET",))

    # 環境変数で上書き可能
    override = os.getenv(f"AGENT_{agent_id.upper()}_MODEL_ID")
    if override:
        return override

    # デフォルトは最優先モデル
    return MODEL_MAP[preferences[0]]

3.3 コスト最適化の具体例

Before(すべて Sonnet 4.5):

  • 100 記事生成 = 100 × (Fact抽出 + 正規化 + 要約 + 執筆 + 校閲) × $0.003 = 約$1.50

After(モデル使い分け):

  • Fact抽出〜要約: Sonnet 4.5 × $0.003
  • 執筆: Sonnet 4.5 × $0.003
  • 校閲: Haiku 4.5 × $0.00025
  • 合計: 約$0.95(37%削減)

4. Rate Limit 制約をアーキテクチャで解決する

4.1 Claude Sonnet 4.5 の制約

  • 2 RPM(Requests Per Minute) = 30秒に1リクエストのみ
  • 連続呼び出しすると ThrottlingException が発生

Rate Limit 制約に対しては、以下の 3 つの解決策を組み合わせます。

4.2 解決策 1: Exponential Backoff with Jitter

import time
import random
from botocore.exceptions import ClientError

def invoke_bedrock_with_retry(
    model_id: str,
    messages: list,
    max_retries: int = 3,
    base_delay: float = 30.0
):
    """Rate Limit対応のリトライロジック"""
    for attempt in range(max_retries):
        try:
            response = bedrock_client.invoke_model(
                modelId=model_id,
                body=json.dumps({"messages": messages})
            )
            return json.loads(response['body'].read())

        except ClientError as e:
            if e.response['Error']['Code'] == 'ThrottlingException':
                if attempt < max_retries - 1:
                    # Exponential Backoff + Jitter
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 5)
                    print(f"Rate limited. Retrying in {delay}s...")
                    time.sleep(delay)
                else:
                    raise
            else:
                raise

4.3 解決策 2: Step Functions による Sequential 実行

# Step Functions定義(一部抜粋)
States:
  ExtractFacts:
    Type: Task
    Resource: arn:aws:lambda:us-east-1:xxx:function:FactExtraction
    Next: Wait30Seconds

  Wait30Seconds:
    Type: Wait
    Seconds: 30  # Rate Limit回避のための待機
    Next: NormalizeFacts

  NormalizeFacts:
    Type: Task
    Resource: arn:aws:lambda:us-east-1:xxx:function:FactNormalization

ポイント:

  • エージェント間に Wait ステート を挿入
  • 30秒の強制待機で Rate Limit を回避
  • 並列実行が不要なワークフローに適用

4.4 解決策 3: キャッシュレイヤーの導入

import hashlib
from functools import lru_cache

class BedrockResponseCache:
    """LLM呼び出し結果をキャッシュ"""

    def __init__(self, s3_bucket: str):
        self.bucket = s3_bucket

    def get_cache_key(self, agent_id: str, payload: dict) -> str:
        """ペイロードのハッシュ値からキャッシュキーを生成"""
        payload_str = json.dumps(payload, sort_keys=True)
        return hashlib.sha256(payload_str.encode()).hexdigest()

    def get(self, agent_id: str, payload: dict) -> dict | None:
        """キャッシュから取得"""
        key = self.get_cache_key(agent_id, payload)
        try:
            obj = s3_client.get_object(
                Bucket=self.bucket,
                Key=f"cache/{agent_id}/{key}.json"
            )
            return json.loads(obj['Body'].read())
        except s3_client.exceptions.NoSuchKey:
            return None

    def put(self, agent_id: str, payload: dict, response: dict):
        """キャッシュに保存"""
        key = self.get_cache_key(agent_id, payload)
        s3_client.put_object(
            Bucket=self.bucket,
            Key=f"cache/{agent_id}/{key}.json",
            Body=json.dumps(response)
        )

効果:

  • 同一ペイロードの再実行時にRate Limit を消費しない
  • デバッグ時の反復テストが高速化

Cache
Cache


5. Agent Runtime Pattern - 薄いラッパー設計

5.1 設計思想

すべてのエージェント(Lambda 関数)は以下の統一インターフェースに従います。

# 例: fact_extraction.py

AGENT_ID = "fact_extraction"

def run(payload: dict | None = None, **overrides) -> dict:
    """エージェント実行のエントリポイント"""
    return invoke_structured_agent(AGENT_ID, payload, **overrides)

def handler(event, _context):
    """Lambda ハンドラー"""
    return run(event)

5.2 共通ランタイムの責務

# 共通ランタイム実装

def invoke_structured_agent(
    agent_id: str,
    payload: dict | None,
    **overrides
) -> dict:
    """すべてのエージェントが共通で使用するランタイム"""

    # 1. プロンプトレジストリから取得
    prompt_config = load_prompt(agent_id)

    # 2. テンプレートレンダリング
    user_message = render_template(
        prompt_config['user_template'],
        payload
    )

    # 3. モデル選択(環境変数 > デフォルト)
    model_id = get_model_for_agent(agent_id)

    # 4. Bedrock呼び出し(リトライ付き)
    response = invoke_bedrock_with_retry(
        model_id=model_id,
        messages=[
            {"role": "system", "content": prompt_config['system_prompt']},
            {"role": "user", "content": user_message}
        ]
    )

    # 5. JSON抽出(最初のJSONブロックをパース)
    structured_output = extract_first_json(response['content'])

    return {
        "raw_text": response['content'],
        "payload": structured_output
    }

5.3 メリット

  • エージェント固有コードが最小化 - AGENT_ID の定義のみ
  • プロンプトとコードの分離 - JSON レジストリで一元管理
  • テストが容易 - モックしやすいシンプルな構造

比較・考察

マルチエージェント設計の選択肢

1. Bedrock Agents (Agent Core) vs Lambda + Step Functions - 詳細比較

マルチエージェントシステムを構築する際、Amazon Bedrock Agents(Agent Core機能を含む)と Lambda + Step Functions のどちらを選ぶべきかは、プロジェクトの要件によって異なります。

1.1 Bedrock Agents (Agent Core) のメリット

Agent Core の主要機能:

  • 自動オーケストレーション: LLM が次のアクションを自律的に決定
  • Knowledge Base 統合: ベクトルストア(Amazon OpenSearch、Aurora)との自動連携
  • Guardrails 統合: コンテンツフィルタリング、PII 保護が組み込み
  • メモリ管理: セッション間での会話履歴の自動保持

メリット:

観点 詳細
開発速度 ✅ コンソール UI で迅速にプロトタイプ構築
✅ オーケストレーションロジックを書く必要なし
マネージド運用 ✅ スケーリング、リトライ、ログが自動管理
✅ インフラ管理不要
自律的な判断 ✅ LLM が状況に応じてツール選択・順序決定
✅ 予期しない入力への柔軟な対応
Knowledge Base ✅ RAG(検索拡張生成)がネイティブ統合
✅ ベクトル検索の実装が不要

デメリット・制約:

観点 詳細
カスタマイズ性 ❌ オーケストレーションフローを細かく制御できない
❌ エラーハンドリングの柔軟性が低い
決定論性 ❌ LLM の判断に依存するため、毎回異なるフローの可能性
❌ 厳密なワークフロー制御が困難
デバッグ ❌ どのエージェントがどの順で実行されたか追跡しづらい
❌ CloudWatch Logs の分散により調査コストが高い
コスト ❌ オーケストレーション用の LLM 呼び出しが増加
❌ Knowledge Base のベクトルストア運用コスト

1.2 Lambda + Step Functions のメリット

主要機能:

  • 明示的なワークフロー定義: ASL(Amazon States Language)による決定論的な制御
  • 柔軟なエラーハンドリング: Retry、Catch、Fallback の細かい設定
  • 可視化: Step Functions コンソールで実行フローを視覚的に追跡
  • 統合性: AWS サービス(DynamoDB、S3、SQS など)との直接統合

メリット:

観点 詳細
完全な制御 ✅ エージェントの実行順序を厳密に定義可能
✅ 条件分岐、並列実行、ループを自由に設計
決定論性 ✅ 同じ入力に対して常に同じフローを保証
✅ テストが容易(再現性が高い)
デバッグ ✅ Step Functions の実行履歴で各ステートを追跡
✅ CloudWatch Logs で Lambda ごとのログを確認
コスト最適化 ✅ 必要な LLM 呼び出しのみを実行
✅ キャッシュレイヤーで重複呼び出しを削減
テスタビリティ ✅ 各 Lambda を独立してユニットテスト可能
✅ Step Functions を Mock した統合テスト

デメリット・制約:

観点 詳細
開発コスト ❌ オーケストレーションロジックを自前で実装
❌ リトライ、エラーハンドリングの設計が必要
運用負荷 ❌ Lambda のデプロイ、バージョン管理が必要
❌ Step Functions の状態遷移設計の学習コスト
柔軟性 ❌ 予期しない入力への対応を事前に設計が必要
❌ 動的なツール選択は自前実装

1.3 本システムで Lambda + Step Functions を選んだ理由

本記事で紹介したシステムでは、以下の理由から Lambda + Step Functions を採用しました。

  1. Fact-First Architecture の要件

    • 事実抽出 → 正規化 → 要約の厳密な順序制御が必須
    • LLM の自律判断による順序変更は許容できない
  2. トレーサビリティ要件

    • どのエージェントがどのデータを生成したか完全に追跡可能である必要
    • Step Functions の実行履歴で監査証跡を確保
  3. コスト効率

    • オーケストレーション用の LLM 呼び出しを削減
    • キャッシュレイヤーによる重複呼び出しの削減
  4. テスト戦略

    • Property-Based Testing で各エージェントの不変条件を検証
    • 決定論的なフローにより再現性の高いテストが可能

1.4 アーキテクチャ選択のディシジョンツリー

以下のフローチャートで、あなたのプロジェクトに最適なアーキテクチャを選択できます。

1.5 ユースケース別の推奨アーキテクチャ

ユースケース 推奨 理由
カスタマーサポートボット Bedrock Agents 自然な対話フロー、Knowledge Base統合
コンテンツ生成パイプライン Lambda + Step Functions 厳密なワークフロー、品質管理
データ分析レポート作成 Lambda + Step Functions 決定論的な処理、監査証跡
社内FAQ検索 Bedrock Agents RAG統合、迅速なプロトタイプ
コード生成・レビュー Lambda + Step Functions テスト可能性、品質保証
ドキュメント要約 Bedrock Agents シンプルなフロー、Knowledge Base活用
金融レポート生成 Lambda + Step Functions トレーサビリティ、コンプライアンス

1.6 Hybrid 構成の可能性

実際のプロジェクトでは、両者を組み合わせる「Hybrid 構成」も有効です。

パターン例:

# Step Functionsで全体フローを制御しつつ、
# 一部のタスクでBedrock Agentを呼び出す

States:
  # 厳密な制御が必要な部分
  ExtractFacts:
    Type: Task
    Resource: arn:aws:lambda:...:function:FactExtraction
    Next: NormalizeFacts

  NormalizeFacts:
    Type: Task
    Resource: arn:aws:lambda:...:function:FactNormalization
    Next: InvokeBedrockAgent

  # 柔軟な対話が必要な部分
  InvokeBedrockAgent:
    Type: Task
    Resource: arn:aws:states:::bedrock:invokeAgent
    Parameters:
      AgentId: "AGENT123"
      AgentAliasId: "ALIAS456"
      InputText.$: "$.normalized_facts"
    Next: FinalizeContent

Hybrid 構成が有効なケース:

  • 初期処理は決定論的に制御、最終調整は LLM の判断に委ねる
  • データ収集は Step Functions、ユーザー対話は Bedrock Agents
  • バッチ処理は Lambda、対話的な問い合わせは Agent

2. Property-Based Testing vs Snapshot Testing

項目 Property-Based Snapshot
非決定性への対応 ◎ 性質をテスト △ 毎回更新が必要
バグ検出力 ◎ エッジケース自動生成 ○ リグレッション検出
学習コスト △ やや高い ◎ 低い

併用パターン:

  • 不変条件 → Property-Based Testing
  • 出力フォーマット → Snapshot Testing

まとめ

本記事で紹介した知見

  1. Fact-First Architecture

    • すべての記述を検証可能な事実に基づかせる
    • 出典 URL・スニペットの必須化
    • Pydantic によるデータモデル検証
  2. Property-Based Testing

    • LLM 出力の不変条件をテスト
    • Hypothesis による自動ケース生成
    • 実運用で発見したバグ事例
  3. Claude 4.5 シリーズの使い分け

    • Sonnet(バランス)、Opus(高精度)、Haiku(高速)
    • 動的モデル選択による 37% コスト削減
  4. Rate Limit 対策

    • Exponential Backoff + Jitter
    • Step Functions の Wait ステート
    • キャッシュレイヤー導入

適用シナリオ

本アーキテクチャが有効な場合:

  • コンテンツのトレーサビリティが求められる(医療・金融・報道など)
  • 複数 LLM を組み合わせた段階的処理が必要
  • AWS Serverless 環境で運用したい

他の選択肢を検討すべき場合:

  • リアルタイム応答が必須(Rate Limit が致命的)
  • AWS 以外のクラウド環境
  • エージェント間の複雑な対話が必要

今後の展開可能性

マルチエージェント型システムでは、以下のような継続的改善が考えられます。

  • 評価レイヤーによる自己改善ループの実装
  • Visual Layer との統合(画像生成エージェント)
  • 多言語対応(English-first → Translation)

参考リンク

Amazon Bedrock

  1. Amazon Bedrock とは - AWS 公式ドキュメント

    • Bedrock の詳細仕様(日本語)
  2. Amazon Bedrock での Anthropic Claude - AWS 公式

    • Claude 4.5 シリーズのモデル一覧(日本語)
  3. Amazon Bedrock Agents とは - AWS 公式

    • Bedrock Agents の概要と使い方(日本語)

Testing

  1. Hypothesis - Property-Based Testing 公式ドキュメント

    • Python 向け Property-Based Testing ライブラリ
  2. Property-based testing とは - GitHub Blog 日本語版

    • Property-Based Testing の基本概念解説

Architecture

  1. AWS Step Functions とは - AWS 公式ドキュメント

    • Step Functions の基本と活用方法(日本語)
  2. AWS Step Functions のベストプラクティス - AWS 公式

    • ワークフロー設計のベストプラクティス(日本語)
  3. AWS Lambda とは - AWS 公式ドキュメント

    • Lambda の基本概念(日本語)

AWS ブログ

  1. Amazon Bedrock に関する記事 - AWS ブログ

    • Bedrock の活用事例とベストプラクティス(日本語)
  2. 生成 AI と機械学習 - AWS ブログ

    • AI/ML 関連の最新情報(日本語)

Discussion