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

課題提起
Amazon Bedrock を活用したマルチエージェント型の LLM システム開発において、以下のような課題に直面していないでしょうか?
- LLM出力の品質保証をどう担保するか - 非決定的な出力をどうテストすべきか
- Rate Limit制約下での設計 - Claude Sonnet 4.5の2 RPM制約にどう対処するか
- トレーサビリティの確保 - 生成コンテンツの信頼性をどう証明するか
- エージェント間の責務分離 - 複雑化するワークフローをどう整理するか
記事の目的
本記事では、マルチエージェント型のコンテンツ生成システムを開発した際に得た、以下の知見を共有します。
- Fact-First Architecture によるトレーサビリティ設計
- Property-Based Testing による LLM 出力の品質保証
- Claude 4.5 シリーズの使い分けパターン(Sonnet/Opus/Haiku)
- 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 関数)が明確な責務を持ち、保守性の高い設計を実現しています。

詳細な解説・分析
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 | 校閲、簡易チェック | 高速・低コスト |

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 を消費しない
- デバッグ時の反復テストが高速化


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 を採用しました。
-
Fact-First Architecture の要件
- 事実抽出 → 正規化 → 要約の厳密な順序制御が必須
- LLM の自律判断による順序変更は許容できない
-
トレーサビリティ要件
- どのエージェントがどのデータを生成したか完全に追跡可能である必要
- Step Functions の実行履歴で監査証跡を確保
-
コスト効率
- オーケストレーション用の LLM 呼び出しを削減
- キャッシュレイヤーによる重複呼び出しの削減
-
テスト戦略
- 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
まとめ
本記事で紹介した知見
-
Fact-First Architecture
- すべての記述を検証可能な事実に基づかせる
- 出典 URL・スニペットの必須化
- Pydantic によるデータモデル検証
-
Property-Based Testing
- LLM 出力の不変条件をテスト
- Hypothesis による自動ケース生成
- 実運用で発見したバグ事例
-
Claude 4.5 シリーズの使い分け
- Sonnet(バランス)、Opus(高精度)、Haiku(高速)
- 動的モデル選択による 37% コスト削減
-
Rate Limit 対策
- Exponential Backoff + Jitter
- Step Functions の Wait ステート
- キャッシュレイヤー導入
適用シナリオ
本アーキテクチャが有効な場合:
- コンテンツのトレーサビリティが求められる(医療・金融・報道など)
- 複数 LLM を組み合わせた段階的処理が必要
- AWS Serverless 環境で運用したい
他の選択肢を検討すべき場合:
- リアルタイム応答が必須(Rate Limit が致命的)
- AWS 以外のクラウド環境
- エージェント間の複雑な対話が必要
今後の展開可能性
マルチエージェント型システムでは、以下のような継続的改善が考えられます。
- 評価レイヤーによる自己改善ループの実装
- Visual Layer との統合(画像生成エージェント)
- 多言語対応(English-first → Translation)
参考リンク
Amazon Bedrock
-
Amazon Bedrock とは - AWS 公式ドキュメント
- Bedrock の詳細仕様(日本語)
-
Amazon Bedrock での Anthropic Claude - AWS 公式
- Claude 4.5 シリーズのモデル一覧(日本語)
-
Amazon Bedrock Agents とは - AWS 公式
- Bedrock Agents の概要と使い方(日本語)
Testing
-
Hypothesis - Property-Based Testing 公式ドキュメント
- Python 向け Property-Based Testing ライブラリ
-
Property-based testing とは - GitHub Blog 日本語版
- Property-Based Testing の基本概念解説
Architecture
-
AWS Step Functions とは - AWS 公式ドキュメント
- Step Functions の基本と活用方法(日本語)
-
AWS Step Functions のベストプラクティス - AWS 公式
- ワークフロー設計のベストプラクティス(日本語)
-
- Lambda の基本概念(日本語)
AWS ブログ
-
Amazon Bedrock に関する記事 - AWS ブログ
- Bedrock の活用事例とベストプラクティス(日本語)
-
- AI/ML 関連の最新情報(日本語)
Discussion