こんにちは、サーバーワークスで生成AIの活用推進を担当している針生です。
以前のブログでAWSのオープンソースSDK「Strands Agents」の概要と利用方法についてご紹介しました。
前回は単一エージェント構成でしたが、実際の業務では複数のエージェントが協調して作業する場面が出てきます。
Strands Agents SDKはマルチエージェント構成のためにAgents-as-Tools、Swarm、Graph、Workflowという4つのパターンを提供していますが、「どのパターンをどのような場面で使うべきか？」という判断が難しいと感じました。
そこで、コードレビューエージェントという同一のユースケースを3パターンで実装し、実行結果を定量・定性の両面から比較してみることにしました。
なお、AWSが提供するマネージドなコードレビューサービスとしてはAWS Security Agentがあり、以前の記事でもご紹介しました。
本記事では、Strands Agents SDKを使って自前でコードレビューエージェントを構築することで、マルチエージェント構成の設計判断を深掘りします。
マルチエージェント構成パターンの概要
まずは、Strands Agents SDKが提供する4つのマルチエージェントパターンを整理します。
| パターン | 実行パスの決定者 | 特徴 | 適するユースケース |
|---|---|---|---|
| Agents-as-Tools | オーケストレーターエージェント | 階層型委任。専門エージェントを「ツール」として呼び出す。オーケストレーターが制御を保持 | 中央管理が必要な場面。専門家への問い合わせ型 |
| Swarm | エージェント同士が自律的に判断 | ハンドオフによる自己組織化。共有メモリで情報共有。実行パスが創発的に決まる | チームワーク型。順序が事前に決まらない協調作業 |
| Graph | 開発者が定義した有向グラフ + エージェントによる条件分岐 | 決定論的な実行順序。サイクル（ループ）もサポート。条件付きエッジで分岐可能 | 承認フローや反復改善など、構造化されたプロセス |
| Workflow | 開発者が定義したタスク依存関係 | 固定パイプライン。依存関係に基づく並列実行 | ETLパイプラインなど、繰り返し実行する定型処理 |
4つのパターンの最も重要な違いは、実行パスを誰が決めるかという点です。Agents-as-Toolsはオーケストレーター、Swarmはエージェント同士の自律判断、Graphは開発者の事前定義、Workflowも開発者の事前定義（固定的）です。
本記事では、設計判断の分かれ目が明確に現れる Agents-as-Tools / Swarm / Graph の3パターンを比較対象とします。Workflowは固定パイプラインであり、コードレビューのような「条件次第で差し戻しが発生する」タスクには適しません。そのため今回は対象外としました。
検証ユースケースの定義
コードレビューを比較対象に選んだのは、マルチエージェントの特性差がはっきり出るユースケースだからです。セキュリティとパフォーマンスという異なる専門性が自然に求められますし、「指摘→修正→再レビュー」の差し戻しサイクルもあるので、パターンごとの制御方式の違いが見えやすくなります。また、意図的に問題を埋め込んだコードを用意すれば「何件検出できたか」で客観的に比較できるのも都合がよいと考えました。
エージェント構成
3つのエージェントで構成するコードレビューシステムを構築します。
セキュリティレビューエージェント： OWASP Top 10に基づく脆弱性チェックを担当。SQLインジェクション、入力バリデーション不足、機密情報のハードコーディングなどを検出します。
パフォーマンスレビューエージェント： AWSサービス利用の効率性を担当。不要なAPI呼び出し、コールドスタートの影響、DynamoDBのスキャン操作の最適化などを分析します。
総合レビューエージェント： 両エージェントの指摘を統合し、深刻度で優先順位をつけ、修正提案を含む最終レビューレポートを生成します。
レビュー対象のサンプルコード
3パターンの比較を行うため、意図的に6つの問題を埋め込んだコードを用意しました。
※ブログでの表示上、コメントを挿入していますが、実際のコードには記述していません。AIエージェントがコメントを検知して動作してしまうためです。
import json import boto3 import pymysql import logging logger = logging.getLogger() logger.setLevel(logging.INFO) # 問題1: 機密情報のハードコーディング DB_HOST = "mydb.cluster-abc123.us-east-1.rds.amazonaws.com" DB_USER = "admin" DB_PASSWORD = "P@ssw0rd123!" DB_NAME = "production_db" def lambda_handler(event, context): user_id = event.get("queryStringParameters", {}).get("user_id") # 問題2: SQLインジェクションの脆弱性 query = f"SELECT * FROM users WHERE user_id = '{user_id}'" # 問題3: Lambda関数内でのDB接続（コールドスタートの度に接続） connection = pymysql.connect( host=DB_HOST, user=DB_USER, password=DB_PASSWORD, database=DB_NAME ) try: with connection.cursor() as cursor: cursor.execute(query) result = cursor.fetchall() # 問題4: ログに機密情報を出力 logger.info(f"Query executed for user: {user_id}, password: {DB_PASSWORD}") # 問題5: DynamoDBの不要なフルスキャン dynamodb = boto3.resource("dynamodb") table = dynamodb.Table("UserActivity") scan_result = table.scan() activities = [ item for item in scan_result["Items"] if item.get("user_id") == user_id ] return { "statusCode": 200, "body": json.dumps({ "user": result, "activities": activities }) } except Exception as e: # 問題6: エラーハンドリングの不足（例外を握りつぶしている） return {"statusCode": 200, "body": json.dumps({"status": "ok"})}
上記のコードには、以下の6つの問題を意図的に埋め込んでいます。
| # | カテゴリ | 問題 |
|---|---|---|
| 1 | セキュリティ | DB接続情報のハードコーディング |
| 2 | セキュリティ | SQLインジェクション脆弱性 |
| 3 | パフォーマンス | Lambda関数内でのDB接続初期化 |
| 4 | セキュリティ | ログへの機密情報出力 |
| 5 | パフォーマンス | DynamoDBの不要なフルスキャン |
| 6 | セキュリティ/品質 | 例外の握りつぶし |
共通条件
3パターンの比較については、以下の共通条件としました。
- 使用モデル：Amazon Bedrock上のClaude Sonnet 4.6（
jp.anthropic.claude-sonnet-4-6） - リージョン：ap-northeast-1（東京）
- 各エージェントのベースとなるシステムプロンプトは3パターンで同一のものを使用（パターン2のみハンドオフ指示を追加）
共通のシステムプロンプト
3つのエージェントには、それぞれ以下のシステムプロンプトを設定しました。
セキュリティレビューエージェント（SECURITY_SYSTEM_PROMPT）
あなたはセキュリティ専門のコードレビュアーです。 OWASP Top 10に基づいて、以下の観点でコードを分析してください。 - SQLインジェクション、XSSなどのインジェクション脆弱性 - 機密情報のハードコーディング - 不適切な認証・認可 - 機密データの不適切なログ出力 - 不十分なエラーハンドリング（セキュリティ上の影響がある場合） 各指摘には以下を含めてください： - 該当コード行（行番号） - 問題の説明 - 深刻度（Critical / High / Medium / Low） - 具体的な修正提案（修正後のコード例を含む）
パフォーマンスレビューエージェント（PERFORMANCE_SYSTEM_PROMPT）
あなたはAWSパフォーマンス専門のコードレビュアーです。 AWSベストプラクティスに基づいて、以下の観点でコードを分析してください。 - Lambda関数のコールドスタート最適化（接続の再利用、初期化の最適化） - DynamoDB操作の効率性（scan vs query、インデックス活用） - 不要なAPI呼び出しの排除 - メモリ・タイムアウト設定の妥当性 - リソースの適切なクリーンアップ 各指摘には以下を含めてください： - 該当コード行（行番号） - 問題の説明 - 深刻度（Critical / High / Medium / Low） - 具体的な修正提案（修正後のコード例を含む）
総合レビューエージェント（COMPREHENSIVE_SYSTEM_PROMPT）
あなたはシニアエンジニアとして、セキュリティとパフォーマンスの
レビュー結果を統合する総合レビュアーです。
以下のタスクを実行してください：
1. セキュリティとパフォーマンスの指摘を統合する
2. 重複する指摘をマージする
3. 深刻度に基づいて優先順位をつける（Critical → High → Medium → Low）
4. 各指摘の修正提案が実装可能か検証する
5. 最終レビューレポートを以下のJSON形式で出力する
出力形式：
{
"findings": [
{
"id": "F001",
"category": "security" | "performance",
"severity": "critical" | "high" | "medium" | "low",
"line": 行番号,
"description": "問題の説明",
"recommendation": "修正提案",
"code_fix": "修正後のコード"
}
],
"summary": {
"total_findings": 件数,
"critical": 件数,
"high": 件数,
"medium": 件数,
"low": 件数,
"overall_risk": "high" | "medium" | "low"
}
}
パターン別実装
準備
Strands Agents SDKのインストールとAmazon Bedrockの設定が完了していることを前提とします。 セットアップ手順は公式ドキュメントをご参照ください。
パターン1：Agents-as-Tools
アーキテクチャ
Agents-as-Toolsパターンでは、総合レビューエージェントがオーケストレーターとなり、セキュリティ・パフォーマンスの各専門エージェントを「ツール」として呼び出す階層型構成を取ります。
この構成では、オーケストレーターが制御を保持し続けます。各エージェントの出力はオーケストレーターに返され、オーケストレーターが次のステップを判断します。エージェント同士は直接やり取りしません。
3パターンの中で最もシンプルな構成なので、まずはここから見ていきます。
実装コード
以下は主要部分の抜粋です。完全なコードはGitHub Gistに公開しています。
from strands import Agent, tool from strands.models import BedrockModel model = BedrockModel( model_id="jp.anthropic.claude-sonnet-4-6", region_name="ap-northeast-1" ) TARGET_CODE = """...""" # 上記のレビュー対象コード # --- セキュリティレビューツール --- @tool def security_review(code: str) -> str: """ コードのセキュリティレビューを実行する。 OWASP Top 10に基づく脆弱性チェックを行い、 指摘事項を返す。 Args: code: レビュー対象のPythonコード Returns: セキュリティレビュー結果 """ security_agent = Agent( model=model, system_prompt=SECURITY_SYSTEM_PROMPT, # 上記のセキュリティ用プロンプト callback_handler=None # オーケストレーターの出力と混ざらないよう抑制 ) response = security_agent(f"以下のコードをセキュリティの観点でレビューしてください:\n\n{code}") return str(response) # --- パフォーマンスレビューツール --- @tool def performance_review(code: str) -> str: """ コードのパフォーマンスレビューを実行する。 AWSベストプラクティスに基づく分析を行い、 指摘事項を返す。 Args: code: レビュー対象のPythonコード Returns: パフォーマンスレビュー結果 """ performance_agent = Agent( model=model, system_prompt=PERFORMANCE_SYSTEM_PROMPT, # 上記のパフォーマンス用プロンプト callback_handler=None ) response = performance_agent(f"以下のコードをパフォーマンスの観点でレビューしてください:\n\n{code}") return str(response) # --- オーケストレーター（総合レビュー）--- orchestrator = Agent( model=model, system_prompt=COMPREHENSIVE_SYSTEM_PROMPT, # 上記の総合レビュー用プロンプト tools=[security_review, performance_review] ) # 実行 result = orchestrator( f"以下のAWS Lambdaハンドラーコードを総合的にレビューしてください。" f"必ずセキュリティレビューとパフォーマンスレビューの両方を実行し、" f"結果を統合してレポートを作成してください。\n\n{TARGET_CODE}" )
実行結果
オーケストレーターはセキュリティレビューとパフォーマンスレビューの両ツールを同時に呼び出し、両方の結果を受け取った後に重複をマージして最終レポートを生成しました。
最終レビューレポート（JSON）
{ "findings": [ { "id": "F001", "category": "security", "severity": "critical", "line": 10, "description": "DBホスト・ユーザー名・パスワード・DB名がソースコードにハードコーディングされている。Gitリポジトリや Lambda コンソールから誰でも閲覧可能。OWASP A02/A07 に該当。", "recommendation": "AWS Secrets Manager にクレデンシャルを移管し、Lambda 環境変数にはシークレット名のみを格納する。", "code_fix": "import os, json, boto3\nfrom functools import lru_cache\n\nSECRET_NAME = os.environ['DB_SECRET_NAME']\n\n@lru_cache(maxsize=1)\ndef get_db_credentials() -> dict:\n client = boto3.client('secretsmanager')\n resp = client.get_secret_value(SecretId=SECRET_NAME)\n return json.loads(resp['SecretString'])" }, { "id": "F002", "category": "security", "severity": "critical", "line": 18, "description": "user_id をそのまま f-string でSQL文字列に埋め込んでおり、SQLインジェクション攻撃が成立する。OWASP A03 に該当。", "recommendation": "pymysql のパラメータ化クエリ（プレースホルダ %s）を使用する。", "code_fix": "query = 'SELECT user_id, username, email FROM users WHERE user_id = %s'\nwith connection.cursor() as cursor:\n cursor.execute(query, (user_id,))\n result = cursor.fetchone()" }, { "id": "F003", "category": "security", "severity": "critical", "line": 32, "description": "logger.info に DB_PASSWORD を直接埋め込んでおり、CloudWatch Logs にパスワードが平文で記録される。OWASP A09 に該当。", "recommendation": "パスワードを含む機密情報を一切ログに含めない。user_id も PII の可能性があるため匿名化する。", "code_fix": "import hashlib\nuid_hash = hashlib.sha256(str(user_id).encode()).hexdigest()[:8]\nlogger.info('Query executed for user_hash: %s', uid_hash)" }, { "id": "F004", "category": "security", "severity": "high", "line": 35, "description": "DynamoDB の table.scan() でテーブル全件を取得しアプリ層でフィルタリングしている。他ユーザーデータがメモリに展開されアクセス制御違反となる。OWASP A01 に該当。", "recommendation": "UserActivity テーブルに user_id を Partition Key とする GSI を作成し、table.query() に置き換える。", "code_fix": "from boto3.dynamodb.conditions import Key\n\n_dynamodb = boto3.resource('dynamodb')\n_activity_table = _dynamodb.Table('UserActivity')\n\ndef get_user_activities(user_id: str) -> list:\n resp = _activity_table.query(\n IndexName='user_id-index',\n KeyConditionExpression=Key('user_id').eq(user_id),\n Limit=100,\n ScanIndexForward=False\n )\n return resp.get('Items', [])" }, { "id": "F005", "category": "security", "severity": "high", "line": 16, "description": "user_id の型・形式・NULL チェックが一切行われていない。None のまま SQL に渡ると意図しないレコードを返す可能性がある。", "recommendation": "正規表現によるホワイトリスト検証を実装し、不正値の場合は即座に 400 Bad Request を返す。", "code_fix": "import re\n\ndef validate_user_id(user_id) -> bool:\n if not user_id:\n return False\n return bool(re.match(r'^[a-zA-Z0-9\\-]{1,36}$', str(user_id)))\n\nparams = event.get('queryStringParameters') or {}\nuser_id = params.get('user_id')\nif not validate_user_id(user_id):\n return {'statusCode': 400, 'body': json.dumps({'error': 'Invalid or missing user_id'})}" }, { "id": "F006", "category": "security", "severity": "high", "line": 43, "description": "全ての例外を捕捉して HTTP 200 / {status: ok} を返している。呼び出し元がエラーを検知できず、例外内容が完全に隠蔽される。OWASP A09 に該当。", "recommendation": "例外を種別ごとに捕捉し適切な HTTP ステータスコード（5xx）を返す。", "code_fix": "except pymysql.MySQLError as e:\n logger.error('Database error: %s', str(e), exc_info=True)\n return {'statusCode': 503, 'body': json.dumps({'error': 'Database unavailable'})}\nexcept Exception as e:\n logger.error('Unexpected error: %s', str(e), exc_info=True)\n return {'statusCode': 500, 'body': json.dumps({'error': 'Internal server error'})}" }, { "id": "F007", "category": "performance", "severity": "high", "line": 19, "description": "pymysql.connect() を lambda_handler 内部で呼び出しているため、Lambda 呼び出しごとに TCP コネクション確立・DB 認証が発生する。100〜300ms のオーバーヘッドとなりコールドスタート悪化の主因となる。", "recommendation": "接続オブジェクトをモジュールレベルのグローバル変数で保持し、コンテナ再利用時は再接続をスキップする。", "code_fix": "from typing import Optional\nimport pymysql\n\n_connection: Optional[pymysql.connections.Connection] = None\n\ndef get_db_connection() -> pymysql.connections.Connection:\n global _connection\n creds = get_db_credentials()\n if _connection is None:\n _connection = pymysql.connect(\n host=creds['host'], user=creds['username'],\n password=creds['password'], database=creds['dbname'],\n cursorclass=pymysql.cursors.DictCursor,\n connect_timeout=5, autocommit=True\n )\n else:\n _connection.ping(reconnect=True)\n return _connection" }, { "id": "F008", "category": "performance", "severity": "medium", "line": 33, "description": "boto3.resource('dynamodb') と dynamodb.Table('UserActivity') を lambda_handler 内部で呼び出しており、Lambda 呼び出しごとにクライアント初期化コストが発生する。", "recommendation": "boto3 クライアント・リソースオブジェクトはモジュールレベルで初期化し、コンテナ再利用時はスキップされる設計とする。", "code_fix": "import os, boto3\n\n_dynamodb = boto3.resource('dynamodb')\n_activity_table = _dynamodb.Table(os.environ['ACTIVITY_TABLE_NAME'])" } ], "summary": { "total_findings": 8, "critical": 3, "high": 4, "medium": 1, "low": 0, "overall_risk": "high" } }
セキュリティとパフォーマンスの重複指摘がマージされ、8件に集約されました。
| ID | カテゴリ | 深刻度 | 指摘概要 |
|---|---|---|---|
| F001 | Security | Critical | DB認証情報のハードコーディング |
| F002 | Security | Critical | SQLインジェクション |
| F003 | Security | Critical | パスワードのCloudWatch Logs平文出力 |
| F004 | Security | High | DynamoDB全件スキャン |
| F005 | Security | High | user_idの型・形式・NULLチェック不在 |
| F006 | Security | High | 全例外をHTTP 200で返却 |
| F007 | Performance | High | 毎回のDB接続確立 |
| F008 | Performance | Medium | boto3クライアントの毎回初期化 |
総合リスク判定：High — 検出 8件（Critical: 3 / High: 4 / Medium: 1）
意図的に埋め込んだ6つの問題の検出状況
| # | 埋め込んだ問題 | 検出 | 対応するFinding |
|---|---|---|---|
| 1 | DB接続情報のハードコーディング | ✓ | F001（Critical） |
| 2 | SQLインジェクション脆弱性 | ✓ | F002（Critical） |
| 3 | Lambda関数内でのDB接続初期化 | ✓ | F007（High） |
| 4 | ログへの機密情報出力 | ✓ | F003（Critical） |
| 5 | DynamoDBの不要なフルスキャン | ✓ | F004（High） |
| 6 | 例外の握りつぶし | ✓ | F006（High） |
6件中6件を検出（検出率100%）。 埋め込んだ問題以外にもF005（入力値の未検証）、F008（boto3クライアントの毎回初期化）の2件を追加で検出しました。追加の2件も妥当な指摘で、実装の手軽さに対して十分な成果だと感じます。
パターン2：Swarm
アーキテクチャ
Swarmパターンは、Agents-as-Toolsとはまったく異なるアプローチです。3つのエージェントが自律的にハンドオフしながらレビューを進めます。
Agents-as-Toolsとの最大の違いは、エージェント同士が共有コンテキストにアクセスしながら自律的にハンドオフする点です。実行パスは事前に決まっておらず、エージェントが自分の担当が完了したと判断したタイミングで次のエージェントに制御を渡します。
実装コード
以下は主要部分の抜粋です。完全なコードはGitHub Gistに公開しています。
from strands.multiagent import Swarm # --- 各エージェントの定義 --- security_agent = Agent( model=model, name="security_reviewer", # Swarmではnameでエージェントを識別 system_prompt=SECURITY_SYSTEM_PROMPT + """ あなたの担当はセキュリティレビューです。 レビューが完了したら、パフォーマンスレビューエージェントまたは 総合レビューエージェントにハンドオフしてください。 パフォーマンスレビューがまだ完了していない場合は、 パフォーマンスレビューエージェントにハンドオフしてください。 """ ) performance_agent = Agent( model=model, name="performance_reviewer", system_prompt=PERFORMANCE_SYSTEM_PROMPT + """ あなたの担当はパフォーマンスレビューです。 レビューが完了したら、セキュリティレビューエージェントまたは 総合レビューエージェントにハンドオフしてください。 セキュリティレビューがまだ完了していない場合は、 セキュリティレビューエージェントにハンドオフしてください。 """ ) comprehensive_agent = Agent( model=model, name="comprehensive_reviewer", system_prompt=COMPREHENSIVE_SYSTEM_PROMPT + """ セキュリティとパフォーマンスの両方のレビューが完了していることを 確認してから、統合レポートを作成してください。 どちらかが不足している場合は、該当するエージェントにハンドオフしてください。 """ ) # --- Swarmの構成 --- swarm = Swarm( nodes=[security_agent, performance_agent, comprehensive_agent], # リストで渡す entry_point=security_agent, # Agentオブジェクトを指定 # ピンポン問題の防止設定 repetitive_handoff_detection_window=8, repetitive_handoff_min_unique_agents=3 ) # 実行 result = swarm( f"以下のAWS Lambdaハンドラーコードを総合的にレビューしてください。\n\n{TARGET_CODE}" )
実行結果
結果的に、実行パスは security_reviewer → performance_reviewer → comprehensive_reviewer と、想定通りの順序で進みました。
最終レビューレポート（JSON）
{ "findings": [ { "id": "F001", "category": "security", "severity": "critical", "line": 17, "description": "SQLインジェクション脆弱性：f-stringによるクエリ文字列の直接構築。user_idに悪意ある入力を与えることで、全ユーザーデータの漏洩・改ざん・削除が可能。", "recommendation": "パラメータ化クエリ（プレースホルダー）を使用する。pymysqlでは%sプレースホルダーとexecute()の第2引数でバインドする。", "code_fix": "query = 'SELECT user_id, name, email FROM users WHERE user_id = %s'\ncursor.execute(query, (user_id,))" }, { "id": "F002", "category": "security", "severity": "critical", "line": 9, "description": "DB認証情報のハードコーディング：ホスト名・ユーザー名・パスワード・DB名がソースコードに平文記載。GitリポジトリやLambdaコンソールから漏洩リスクがあり、ローテーションも不可能。", "recommendation": "AWS Secrets ManagerまたはSystems Manager Parameter Store（SecureString）に認証情報を移行する。", "code_fix": "import boto3\nimport json\n\nsecretsmanager = boto3.client('secretsmanager')\n\ndef get_db_credentials():\n secret = secretsmanager.get_secret_value(SecretId='prod/mydb/credentials')\n return json.loads(secret['SecretString'])" }, { "id": "F003", "category": "security", "severity": "critical", "line": 32, "description": "機密情報のログ出力：DB_PASSWORDがCloudWatch Logsに平文で記録される。ログへのアクセス権を持つ全員がパスワードを取得可能。", "recommendation": "ログ文からDB_PASSWORDを削除する。認証情報・個人情報・セッショントークン等は絶対にログ出力しない。", "code_fix": "logger.info(f'Query executed for user_id: {user_id}')" }, { "id": "F004", "category": "performance", "severity": "critical", "line": 19, "description": "DBコネクションの毎回生成：lambda_handler内でpymysql.connect()を毎回呼び出しており、コネクション確立のオーバーヘッド（数百ms）が全リクエストで発生。ウォームスタートの最大メリットを喪失。", "recommendation": "コネクションをグローバルスコープで管理し、ウォームスタート時に再利用するget_connection()パターンを採用する。", "code_fix": "_connection = None\n\ndef get_connection():\n global _connection\n creds = get_db_credentials()\n if _connection is None or not _connection.open:\n _connection = pymysql.connect(\n host=creds['host'], user=creds['username'],\n password=creds['password'], database=creds['dbname'],\n connect_timeout=5, read_timeout=10, write_timeout=10,\n cursorclass=pymysql.cursors.DictCursor\n )\n return _connection" }, { "id": "F005", "category": "security", "severity": "critical", "line": 34, "description": "【セキュリティ×パフォーマンス複合問題】DynamoDB全件スキャン：table.scan()でテーブル全件取得後にPython側でuser_idフィルタリング。他ユーザーの全データがLambdaメモリに展開されプライバシー侵害リスク。データ量増加でRCU消費・レイテンシが線形増大。", "recommendation": "user_id-indexというGSIを作成し、table.query()で該当ユーザーのデータのみ取得する。", "code_fix": "dynamodb = boto3.resource('dynamodb')\ntable = dynamodb.Table('UserActivity')\n\nresponse = table.query(\n IndexName='user_id-index',\n KeyConditionExpression=boto3.dynamodb.conditions.Key('user_id').eq(user_id)\n)\nactivities = response['Items']" }, { "id": "F006", "category": "security", "severity": "high", "line": 15, "description": "入力バリデーションと認可の欠如：user_idの型・形式・長さの検証がなく、認証トークン検証も存在しない。任意のuser_idで他ユーザー情報にアクセス可能なBOLA脆弱性。", "recommendation": "user_idのバリデーションと、JWTやCognito Authorizerによる認証・認可を実装する。", "code_fix": "import re\n\ndef validate_user_id(user_id):\n if not user_id:\n raise ValueError('user_id is required')\n if not re.match(r'^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$', user_id, re.IGNORECASE):\n raise ValueError('Invalid user_id format')\n return user_id" }, { "id": "F007", "category": "performance", "severity": "high", "line": 34, "description": "boto3リソースの毎回初期化：dynamodb = boto3.resource()とtable = dynamodb.Table()がハンドラー内に存在し、ウォームスタート時も毎回実行される。", "recommendation": "boto3.resource()とTable()の初期化をグローバルスコープに移動する。", "code_fix": "dynamodb = boto3.resource('dynamodb')\ntable = dynamodb.Table('UserActivity')" }, { "id": "F008", "category": "performance", "severity": "high", "line": 19, "description": "コネクションリーク：connection.close()がfinallyブロックに存在せず、例外発生時にRDS接続がリリースされない。Lambda同時実行数増加時にRDS接続数が枯渇するリスク。", "recommendation": "finallyブロックでconnection.close()を確実に実行する。", "code_fix": "try:\n with connection.cursor() as cursor:\n cursor.execute(query, (user_id,))\n result = cursor.fetchall()\nexcept Exception as e:\n logger.error(f'DB error: {e}', exc_info=True)\n raise" }, { "id": "F009", "category": "security", "severity": "medium", "line": 41, "description": "エラー隠蔽：全例外をキャッチしてHTTP 200・{'status': 'ok'}を返している。エラーが運用監視で検知不能になる。", "recommendation": "例外をログに記録し、適切なHTTPステータスコード（4xx/5xx）を返す。", "code_fix": "except pymysql.Error as e:\n logger.error(f'Database error: {e}', exc_info=True)\n return {'statusCode': 500, 'body': json.dumps({'error': 'Internal Server Error'})}" }, { "id": "F010", "category": "performance", "severity": "medium", "line": 17, "description": "SELECT *による不要データ転送：usersテーブルの全カラムを取得。不要なネットワーク転送・メモリ消費が発生。", "recommendation": "必要なカラムのみを明示的に指定する。", "code_fix": "query = 'SELECT user_id, name, email, created_at FROM users WHERE user_id = %s'" }, { "id": "F011", "category": "performance", "severity": "medium", "line": 19, "description": "DB接続タイムアウト未設定：connect_timeout/read_timeout/write_timeoutが未設定。DNS障害時にLambdaのタイムアウト上限まで待機し続けるリスク。", "recommendation": "pymysql.connect()に適切なタイムアウト値を設定する。", "code_fix": "pymysql.connect(\n connect_timeout=5, read_timeout=10, write_timeout=10\n)" }, { "id": "F012", "category": "performance", "severity": "low", "line": null, "description": "RDS Proxy未使用：Lambdaスケールアウト時にRDSへの接続数が同時実行数分だけ増加し、接続数上限を超えるリスク。", "recommendation": "Lambda → RDS Proxy → RDS Aurora の構成を導入する。", "code_fix": "# インフラ変更：RDS Proxyを作成し、DB_HOSTをProxyエンドポイントに変更" } ], "summary": { "total_findings": 12, "critical": 5, "high": 3, "medium": 3, "low": 1, "overall_risk": "high" } }
Agents-as-Toolsの8件に対して12件と、共有コンテキストの効果が件数にも現れています。
| ID | カテゴリ | 深刻度 | 指摘概要 |
|---|---|---|---|
| F001 | Security | Critical | SQLインジェクション |
| F002 | Security | Critical | DB認証情報のハードコーディング |
| F003 | Security | Critical | パスワードのCloudWatch Logs平文出力 |
| F004 | Performance | Critical | DBコネクションの毎回生成 |
| F005 | Security/Performance | Critical | DynamoDB全件スキャン |
| F006 | Security | High | 入力バリデーション・認可の欠如 |
| F007 | Performance | High | boto3リソースの毎回初期化 |
| F008 | Performance | High | コネクションリーク |
| F009 | Security | Medium | エラー隠蔽 |
| F010 | Performance | Medium | SELECT *による不要データ転送 |
| F011 | Performance | Medium | DB接続タイムアウト未設定 |
| F012 | Performance | Low | RDS Proxy未使用 |
総合リスク判定：High — 検出 12件（Critical: 5 / High: 3 / Medium: 3 / Low: 1）
意図的に埋め込んだ6つの問題の検出状況：
| # | 埋め込んだ問題 | 検出 | 対応するFinding |
|---|---|---|---|
| 1 | DB接続情報のハードコーディング | ✓ | F002（Critical） |
| 2 | SQLインジェクション脆弱性 | ✓ | F001（Critical） |
| 3 | Lambda関数内でのDB接続初期化 | ✓ | F004（Critical） |
| 4 | ログへの機密情報出力 | ✓ | F003（Critical） |
| 5 | DynamoDBの不要なフルスキャン | ✓ | F005（Critical） |
| 6 | 例外の握りつぶし | ✓ | F009（Medium） |
6件中6件を検出（検出率100%）。 埋め込んだ問題以外にもF006（入力バリデーション・認可欠如）、F007（boto3毎回初期化）、F008（コネクションリーク）、F010（SELECT *）、F011（タイムアウト未設定）、F012（RDS Proxy未使用）の6件を追加で検出しました。
パターン3：Graph
アーキテクチャ
Graphパターンは、3つの中で最も開発者の設計意図を反映できるパターンです。ノード（エージェント）とエッジ（依存関係・条件分岐）を事前に定義し、並列実行・条件分岐・ループを明示的に設計します。その分、実装量は最も多くなります。
実装コード
以下は主要部分の抜粋です。完全なコードはGitHub Gistに公開しています。
from strands.multiagent import GraphBuilder # --- 各エージェントの定義 --- security_agent = Agent( model=model, system_prompt=SECURITY_SYSTEM_PROMPT, callback_handler=None # 総合レビューの出力と混ざらないよう抑制 ) performance_agent = Agent( model=model, system_prompt=PERFORMANCE_SYSTEM_PROMPT, callback_handler=None ) comprehensive_agent = Agent( model=model, system_prompt=COMPREHENSIVE_SYSTEM_PROMPT ) # --- GraphBuilderでグラフを構成 --- builder = GraphBuilder() # ノードの追加 builder.add_node(security_agent, "security_reviewer") builder.add_node(performance_agent, "performance_reviewer") builder.add_node(comprehensive_agent, "comprehensive_reviewer") # エッジの定義 # 両方の結果が揃ったら総合レビューに渡される builder.add_edge("security_reviewer", "comprehensive_reviewer") builder.add_edge("performance_reviewer", "comprehensive_reviewer") # エントリポイントの設定（両方を指定することで並列実行される） builder.set_entry_point("security_reviewer") builder.set_entry_point("performance_reviewer") # 条件付きエッジ：重大問題があれば再レビュー（最大2回まで） _re_review_count = 0 _max_re_reviews = 2 def should_re_review(state) -> bool: """総合レビューの出力に基づいて再レビューが必要か判定（最大2回）""" global _re_review_count if _re_review_count >= _max_re_reviews: return False # 再レビュー回数の上限に達した comprehensive_result = state.results.get("comprehensive_reviewer") if comprehensive_result and comprehensive_result.result: output = str(comprehensive_result.result) if '"critical": 0' not in output.lower(): _re_review_count += 1 return True return False builder.add_edge("comprehensive_reviewer", "security_reviewer", condition=should_re_review) # 実行上限の設定 builder.set_max_node_executions(10) builder.reset_on_revisit(True) # 再訪問時にエージェントの状態をリセット graph = builder.build() # 実行 result = graph( f"以下のAWS Lambdaハンドラーコードをレビューしてください。\n\n{TARGET_CODE}" )
実行結果
Graphパターンでは、セキュリティレビューとパフォーマンスレビューが並列に実行され、両方の完了後に総合レビューが統合処理を行いました。ここまでは想定通りですが、条件付きエッジにより「Critical指摘が残っているなら再レビュー」というサイクルが自動的に発動しました。
実行フローの詳細
実測のノード実行順序は以下の通りです。
- security_reviewer（初回）
- performance_reviewer（初回、1と並列）
- comprehensive_reviewer（1回目）
- security_reviewer（再レビュー）
- comprehensive_reviewer（2回目・最終）
最終レビューレポート（JSON）
{ "findings": [ { "id": "F001", "category": "security", "severity": "critical", "line": 17, "description": "SQLインジェクション脆弱性：ユーザー入力user_idを文字列結合で直接SQLクエリに埋め込んでいる。OWASP A03:2021に該当。SELECT *使用によりパスワードハッシュ等の機密カラムも漏洩する。", "recommendation": "パラメータ化クエリを使用し、必要カラムのみ明示的に指定する。", "code_fix": "query = 'SELECT user_id, name, email, created_at FROM users WHERE user_id = %s'\ncursor.execute(query, (user_id,))" }, { "id": "F002", "category": "security", "severity": "critical", "line": 10, "description": "機密情報のハードコーディング：DB_PASSWORDがソースコードに平文記述。Gitリポジトリ履歴に永続的に残存。adminユーザー使用は最小権限原則にも違反。F003との複合でソースコード・ログの2経路で漏洩。", "recommendation": "AWS Secrets Managerに移行し、lru_cacheでウォーム実行時の取得コストを削減する。", "code_fix": "import os, json\nfrom functools import lru_cache\n\n@lru_cache(maxsize=1)\ndef get_db_credentials() -> dict:\n client = boto3.client('secretsmanager')\n secret = client.get_secret_value(SecretId=os.environ['DB_SECRET_ARN'])\n return json.loads(secret['SecretString'])" }, { "id": "F003", "category": "security", "severity": "critical", "line": 31, "description": "機密情報のログ出力：DB_PASSWORDがCloudWatch Logsに平文で永続的に記録される。OWASP A09:2021に該当。user_idも個人識別情報でありマスキングが推奨される。", "recommendation": "パスワードをログから完全除去し、user_idはマスキング処理を適用する。", "code_fix": "masked_id = f'{str(user_id)[:4]}****' if user_id and len(str(user_id)) > 4 else '****'\nlogger.info(f'Query executed for user_id: {masked_id}')" }, { "id": "F004", "category": "security", "severity": "high", "line": 33, "description": "DynamoDB全件スキャンによるアクセス制御不備とパフォーマンス問題：table.scan()で全ユーザーデータをメモリ展開。OWASP A01:2021に該当。ページネーション未実装で1MB超データが無言で欠落。", "recommendation": "GSI（user_id-index）を作成しQuery APIでサーバー側フィルタリング。ページネーションも実装する。", "code_fix": "from boto3.dynamodb.conditions import Key\n\ndef get_user_activities(table, user_id: str) -> list:\n activities = []\n query_kwargs = {\n 'IndexName': 'user_id-index',\n 'KeyConditionExpression': Key('user_id').eq(user_id),\n 'Limit': 100\n }\n while True:\n response = table.query(**query_kwargs)\n activities.extend(response['Items'])\n if 'LastEvaluatedKey' not in response:\n break\n query_kwargs['ExclusiveStartKey'] = response['LastEvaluatedKey']\n return activities" }, { "id": "F005", "category": "security", "severity": "high", "line": 42, "description": "不適切なエラーハンドリングによる障害・攻撃の隠蔽：全例外を握りつぶしHTTP 200を返却。OWASP A09:2021に該当。エラー内容がログに記録されず障害調査・セキュリティ監視が不能。", "recommendation": "例外種別ごとに適切なHTTPステータスコードを返し、エラー詳細はログにのみ記録する。", "code_fix": "except pymysql.OperationalError as e:\n logger.error(f'DB接続エラー: {type(e).__name__}', exc_info=True)\n return {'statusCode': 503, 'body': json.dumps({'error': 'Service temporarily unavailable'})}\nexcept Exception as e:\n logger.error(f'予期しないエラー: {type(e).__name__}', exc_info=True)\n return {'statusCode': 500, 'body': json.dumps({'error': 'Internal server error'})}" }, { "id": "F006", "category": "performance", "severity": "high", "line": 19, "description": "DB接続のコールドスタート最適化不足とリソースリーク：ハンドラー内で毎回新規TCP接続を確立。boto3クライアントも毎回初期化。finallyブロックがなく接続リーク。タイムアウト未設定。", "recommendation": "DB接続とboto3クライアントをグローバルスコープで管理し、接続状態チェックで切断時のみ再接続する。", "code_fix": "dynamodb = boto3.resource('dynamodb')\nactivity_table = dynamodb.Table('UserActivity')\n_db_connection = None\n\ndef get_db_connection():\n global _db_connection\n if _db_connection is None or not _db_connection.open:\n credentials = get_db_credentials()\n _db_connection = pymysql.connect(\n host=DB_HOST, user=credentials['username'],\n password=credentials['password'], database=DB_NAME,\n connect_timeout=5, read_timeout=10, write_timeout=10,\n charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor\n )\n return _db_connection" }, { "id": "F007", "category": "security", "severity": "medium", "line": 16, "description": "入力バリデーションの欠如：user_idがNoneでもクエリ実行へ進む。queryStringParametersがNoneの場合の考慮も不足。", "recommendation": "user_idの存在確認とフォーマット検証を実装し、不正な場合はHTTP 400で早期リターンする。", "code_fix": "import re\n\ndef validate_user_id(user_id) -> str:\n if not user_id:\n raise ValueError('user_id is required')\n if not re.match(r'^[a-zA-Z0-9_-]{1,64}$', user_id):\n raise ValueError('Invalid user_id format')\n return user_id" }, { "id": "F008", "category": "performance", "severity": "medium", "line": 17, "description": "SELECT *の使用によるデータ過剰取得：不要なネットワーク転送量が発生し、パスワードハッシュ等の機密カラムがレスポンスに含まれるリスクもある。", "recommendation": "必要なカラムのみを明示的に指定する。", "code_fix": "query = 'SELECT user_id, name, email, created_at FROM users WHERE user_id = %s'" }, { "id": "F009", "category": "performance", "severity": "low", "line": 19, "description": "タイムアウト・接続設定の欠如：connect_timeout/read_timeout/write_timeout未設定。charset未設定によるマルチバイト文字化けリスク。", "recommendation": "タイムアウト・文字コード・カーソルクラスを明示的に設定する。", "code_fix": "pymysql.connect(connect_timeout=5, read_timeout=10, write_timeout=10, charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor)" }, { "id": "F010", "category": "performance", "severity": "low", "line": 40, "description": "レスポンスのJSONシリアライズ非対応：datetime/Decimal型でTypeErrorが発生する可能性。F005のエラー隠蔽と組み合わさると障害が表面化しない。", "recommendation": "カスタムJSONEncoderを実装し、datetime→isoformat()、Decimal→float()に変換する。", "code_fix": "class CustomJSONEncoder(json.JSONEncoder):\n def default(self, obj):\n if isinstance(obj, (datetime, date)):\n return obj.isoformat()\n if isinstance(obj, Decimal):\n return float(obj)\n return super().default(obj)" } ], "summary": { "total_findings": 10, "critical": 3, "high": 3, "medium": 2, "low": 2, "overall_risk": "high" } }
再レビューサイクルを経て、関連する指摘が統合された最終的な10件です。
| ID | カテゴリ | 深刻度 | 指摘概要 |
|---|---|---|---|
| F001 | Security | Critical | SQLインジェクション |
| F002 | Security | Critical | DB認証情報のハードコーディング |
| F003 | Security | Critical | パスワードのCloudWatch Logs平文出力 |
| F004 | Security | High | DynamoDB全件スキャン |
| F005 | Security | High | 不適切なエラーハンドリング |
| F006 | Performance | High | DB接続のコールドスタート最適化不足 |
| F007 | Security | Medium | 入力バリデーションの欠如 |
| F008 | Performance | Medium | SELECT *による不要データ転送 |
| F009 | Performance | Low | タイムアウト・接続設定の欠如 |
| F010 | Performance | Low | JSONシリアライズ非対応 |
総合リスク判定：High — 検出 10件（Critical: 3 / High: 3 / Medium: 2 / Low: 2）
意図的に埋め込んだ6つの問題の検出状況：
| # | 埋め込んだ問題 | 検出 | 対応するFinding |
|---|---|---|---|
| 1 | DB接続情報のハードコーディング | ✓ | F002（Critical） |
| 2 | SQLインジェクション脆弱性 | ✓ | F001（Critical） |
| 3 | Lambda関数内でのDB接続初期化 | ✓ | F006（High） |
| 4 | ログへの機密情報出力 | ✓ | F003（Critical） |
| 5 | DynamoDBの不要なフルスキャン | ✓ | F004（High） |
| 6 | 例外の握りつぶし | ✓ | F005（High） |
6件中6件を検出（検出率100%）。 さらに、埋め込んだ問題以外にもF007（入力バリデーション欠如）、F008（SELECT *）、F009（タイムアウト未設定）、F010（JSONシリアライズ非対応）の4件を追加で検出しました。
検証結果の比較
定量比較
同一のLambdaハンドラーコード（意図的に6つの問題を埋め込み）で各パターンを3回ずつ実行した結果の平均値です。
| 評価項目 | Agents-as-Tools | Swarm | Graph |
|---|---|---|---|
| 総実行時間（3回平均） | 150.7秒 | 184.1秒 | 221.4秒 |
| 総トークン消費量（3回平均） | 33,166 | 39,497 | 64,206 |
| エージェント実行回数 | 3回（セキュリティ1 + パフォーマンス1 + 総合1） | 3回（セキュリティ1 + パフォーマンス1 + 総合1） | 5回（セキュリティ2 + パフォーマンス1 + 総合2）※再レビュー最大2回の制限により停止 |
| エージェント間のやり取り形態 | ツール呼び出し2回（並列） | ハンドオフ2回（逐次） | エッジ遷移4回（並列＋再レビューサイクル1回） |
| 問題検出数（既知6件中） | 6/6件 | 6/6件 | 6/6件 |
| 追加検出数（既知6件以外） | 2件 | 6件 | 4件 |
| 総指摘件数（マージ後） | 8件 | 12件 | 10件 |
| 深刻度分布（C/H/M/L） | 3/4/1/0 | 5/3/3/1 | 3/3/2/2 |
計測方法について： 実行時間はPythonのtime.time()による壁時計時間です。トークン消費量はStrands Agents SDKの組み込みメトリクスを利用しています。SwarmとGraphでは、結果オブジェクト（SwarmResult / GraphResult）のaccumulated_usageが全エージェントのトークンを自動集計します。一方、Agents-as-ToolsではオーケストレーターのAgentResult.metrics.accumulated_usageにサブエージェント（@tool内で生成されるAgent）のトークンが含まれないため、ツール関数内でサブエージェントのメトリクスを手動収集し、オーケストレーター分と合算しています。
定量比較から見えたこと
実行時間とコスト： Agents-as-Toolsが平均150.7秒で最速、Swarmが184.1秒、Graphが221.4秒でした。トークン消費量はAgents-as-Toolsの33,166トークンに対し、Swarmが39,497トークン（約1.2倍）、Graphが64,206トークン（約1.9倍）でした。Graphは再レビューサイクル（最大2回）により他パターンより多くのトークンを消費しますが、条件付きエッジにカウンターを設けて回数を制限することで、コストを実用的な範囲に抑えられています。
計測値のブレ： Agents-as-ToolsとGraphは3回の計測で安定した結果を示しました。一方、Swarmは3回の計測で36,035〜44,752トークンと約24%のブレが見られました。これはSwarmの非決定論的な性質（ハンドオフの判断がLLMに委ねられるため、各エージェントの出力量が毎回異なる）に起因します。本番運用でコストを見積もる場面ではこの不確実性がネックになりそうです。
なお、Swarmのハンドオフパスを3回の計測で確認したところ、実行パス自体は3回とも同一でした。
| 実行 | ハンドオフパス | ハンドオフ回数 | 総トークン |
|---|---|---|---|
| Run 1 | security_reviewer → performance_reviewer → comprehensive_reviewer | 2回 | 36,035 |
| Run 2 | security_reviewer → performance_reviewer → comprehensive_reviewer | 2回 | 37,705 |
| Run 3 | security_reviewer → performance_reviewer → comprehensive_reviewer | 2回 | 44,752 |
パスの再現性は高い一方、トークン消費量は36,035〜44,752と約24%の幅がありました。つまり、今回のケースではSwarmのブレは「実行パスの違い」ではなく「同じパスでもLLMの出力量が変動する」ことが主因です。
問題検出数： 3パターンとも意図的に埋め込んだ6件すべてを検出しました（検出率100%）。ただし、追加検出数には明確な差が出ました。Swarmが最多の6件を追加検出した一方、Agents-as-Toolsは2件にとどまりました。Swarmでは共有コンテキストにより各エージェントが他の観点も踏まえた分析を行えるため、「コネクションリーク」「SELECT *」「タイムアウト未設定」「RDS Proxy未使用」といった複合的な指摘が生まれやすい傾向が見られました。
指摘の質と深刻度分布： Swarmは検出数が最多（12件）ですが、深刻度の分布はCritical 5件と他パターンより高めに評価される傾向がありました。これは共有コンテキストでセキュリティ・パフォーマンス双方の文脈が混ざり、深刻度が「積み上がる」効果があるためと考えられます。一方、Graphでは再レビューサイクルを通じて関連する問題が統合・整理され、深刻度分布が3/3/2/2とバランスの取れた結果になりました。
コスト効率の総合評価： 「1件あたりのトークン消費量」で比較すると、Swarmが最も効率的です（39,497トークンで12件 = 約3,291トークン/件）。Agents-as-Toolsも効率が良く（33,166トークンで8件 = 約4,146トークン/件）。Graphは64,206トークンで10件 = 約6,421トークン/件と他パターンより多いですが、再レビューサイクルによる指摘の統合・洗練を考慮すると、許容できる範囲です。
定性比較
| 観点 | Agents-as-Tools | Swarm | Graph |
|---|---|---|---|
| 実行パスの予測可能性 | 高い（オーケストレーター制御） | 低い（創発的に決まる） | 高い（開発者が定義） |
| 柔軟性 | 中（オーケストレーターの判断力に依存） | 高い（自律協調で想定外のパスも取れる） | 中（事前定義が必要） |
| 並列実行 | 可能（parallel tool use） | 不可（ハンドオフは逐次） | 可能（依存関係のないノード） |
| デバッグのしやすさ | 高（ツール呼び出しログが明確） | 低（ハンドオフの理由が不透明） | 高（グラフ構造で可視化可能） |
| エラーハンドリング | 中（オーケストレーターで集約） | 低（自律的なため制御しにくい） | 高（ノードごとに定義可能） |
| コード量 | 少ない（@toolデコレーターのみ） | 中（ハンドオフ条件の記述が必要） | 多い（ノード・エッジの定義が必要） |
| 再現性 | 高い | 低い | 高い |
| 修正提案の具体性 | 高（実用的だが指摘が個別） | 中（検出数は多いが関連性の整理が不十分） | 高（再レビューで関連指摘が統合され、そのまま適用可能な修正コードに） |
各パターンで発生した問題と解決策まとめ
Agents-as-Tools
今回の実行では特に問題なく動きました。オーケストレーターが2つのツールを並列に呼び出し、結果を統合するという想定通りの動作でした。
Swarm
今回の実行では大きな問題は発生しませんでした。しかし、Swarmパターンでは以下の問題が一般的に知られているため注意点としてまとめておきます。
ピンポン問題： 2エージェント間でハンドオフが繰り返されるループです。repetitive_handoff_detection_windowとrepetitive_handoff_min_unique_agentsを設定することで防止できます。今回の実装コードでも設定しています。
役割の越境： 共有コンテキストを通じて他エージェントの情報にアクセスできるため、他エージェントの担当範囲まで指摘してしまう問題です。システムプロンプトで担当範囲を明示することで緩和できます。
ハンドオフタイミングの不安定さ： 十分な分析を行う前にハンドオフする、またはパフォーマンスレビューをスキップして総合レビューに直接ハンドオフするなど、想定外の実行パスが発生しうる問題です。ハンドオフの判断基準や完了確認条件をシステムプロンプトに明記することで改善できます。
Graph
Graphパターンでは以下の問題が発生していました。
当初、再レビューサイクルが上限まで回り続けました。最初の実装では条件付きエッジに回数制限を設けておらず、Critical指摘が残り続ける限り再レビューが繰り返されました。max_node_executions=10で停止はしたものの、セキュリティレビューが5回、総合レビューが4回実行され、トークン消費量が他パターンの約8.7倍に膨らみました。
対策： 条件付きエッジにカウンターを追加し、再レビューを最大2回に制限しました。これにより、ノード実行回数は10回→5回（セキュリティ2 + パフォーマンス1 + 総合2）に削減され、トークン消費量も約1.9倍と実用的な範囲に収まりました。Graphパターンで条件付きエッジを使用する場合、max_node_executionsはセーフティネットとして設定しつつ、条件付きエッジ自体に回数制限を組み込むことを推奨します。
判断フローチャート
3パターンを実際に実装してみた経験から、パターン選択の判断フローチャートを作成してみました。
コードレビューにおける推奨
個人的な結論を先に言うと、まずAgents-as-Toolsで始めて、必要になったらGraphに移行するのが最も現実的だと考えています。
Agents-as-Toolsは実装が最もシンプルで、トークン消費量も最小。埋め込んだ6件の問題をすべて検出でき、修正提案も実用的でした。「まず動くものを作る」第一歩として最適です。
Graphは、差し戻しサイクルが要件に含まれる場合に威力を発揮します。再レビューで指摘が統合・洗練される体験は、他の2パターンにない価値です。ただし、条件付きエッジの回数制限を入れ忘れるとループが発生したので、設計時に注意が必要です。
Swarmは検出の網羅性が最も高く、共有コンテキストによる相互作用は興味深い結果でした。しかし、コストのブレ（約24%）と実行パスの非決定性は、本番運用ではリスクになります。「チーム内のブレストを自動化したい」のような、結果の再現性を求めない用途には向いているかもしれません。
まとめ
検出能力自体は3パターンとも同等で、埋め込んだ6件をすべて検出しました。差が出たのは「追加でどこまで見つけるか」と「指摘をどこまで整理するか」という部分です。Swarmは共有コンテキストの相互作用で追加検出が最多になり、Graphは再レビューサイクルで関連する指摘を統合した質の高いレポートを出しました。ただし、Graphの再レビュー機構は回数制限を入れ忘れるとコストが膨らむという落とし穴がありました。
コスト面では、Agents-as-Toolsのトータルトークン量に対してSwarmが約1.2倍、Graphが約1.9倍。Swarmのコストのブレは、本番で見積もりを出す場面では気になるレベルです。
結局のところ、「どのパターンが最適か」ではなく「自分のユースケースでどの特性が必要か」がパターン選択の判断基準になります。迷ったらAgents-as-Toolsから始めて、差し戻しが必要になったらGraphへ、自律的な協調が欲しくなったらSwarmへ移行するのが、手戻りの少ない進め方だと思います。
今回はコードレビューという比較的わかりやすいユースケースでしたが、パターンの組み合わせ（例えば、GraphのノードにAgents-as-Toolsを組み込む）も試してみたいところです。
おまけ：モデルを Opus 4.6 に変えたらどうなる？
本記事の検証はすべてClaude Sonnet 4.6で実施しましたが、上位モデルのClaude Opus 4.6に変えた場合にどうなるか、Agents-as-Toolsパターンで試してみました。コードの変更はモデルIDの1行のみです。
model = BedrockModel(
model_id="global.anthropic.claude-opus-4-6-v1", # Sonnet → Opus に変更
region_name="ap-northeast-1",
)
| 指標 | Sonnet 4.6（3回平均） | Opus 4.6（3回平均） |
|---|---|---|
| 総実行時間 | 150.7秒 | 161.6秒 |
| 総トークン消費量 | 33,166 | 29,903 |
| 検出件数 | 8件 | 10件 |
実行時間はほぼ同等ですが、トークン消費量はOpusの方が約10%少なく、検出件数は8件→10件に増えました。
ただし、Opusはモデル料金がSonnetより高いため、トークン消費量が少なくても総コストは増加する点に注意が必要です。
参考リンク
- Strands Agents SDK 公式ドキュメント: https://strandsagents.com/
- Strands Agents GitHub: https://github.com/strands-agents/sdk-python
- Multi-Agent Patterns ドキュメント: https://strandsagents.com/docs/user-guide/concepts/multi-agent/multi-agent-patterns/
- AWS公式ブログ「Introducing Strands Agents 1.0」: https://aws.amazon.com/blogs/opensource/introducing-strands-agents-1-0-production-ready-multi-agent-orchestration-made-simple/
- AWS公式ブログ「Multi-Agent collaboration patterns with Strands Agents and Amazon Nova」: https://aws.amazon.com/blogs/machine-learning/multi-agent-collaboration-patterns-with-strands-agents-and-amazon-nova/
針生 泰有(執筆記事の一覧)
サーバーワークスで生成AIの活用推進を担当
