どうもこんにちは。
以前、「Strands AgentsのWorkflowツールを試してみた」というタイトルの記事を書きました。
ここでは、タスク分解したワークフローに沿ってサブエージェントが順番に実行する という検証を行いました。
今回は、Strands AgentsのGraphBuilderを使って、複数の専門エージェントに技術選定レビューを分担させる検証をしてみました。
前回はstrands_tools.workflowを使いましたが、今回はStrands SDK本体にあるGraphを使います。
正直、ドキュメントを読んでいる時、Workflowとgraphの違いはピンときていなかったので、自分にとっては良い検証になりました。
題材は同じく「技術選定&調査エージェント」です。
何か新しいサービスを開発する時に調査を依頼したり、自分の頭の中に浮かんでいる構成を評価してもらうための「技術選定&調査エージェント」です。
たとえば、次のような入力を渡します。
{
"prompt": "新規B2B SaaSを、Next.js + FastAPI + Aurora PostgreSQL + ECS Fargate + Bedrockで作る案を評価して"
}
この時、想定している動作としては、以下の観点に分けてレビューし、最後に採用判断つきのMarkdownレポートを返す。ということを想定しています。（前回の記事と同様ですね。）
- 要件整理
- フロントエンド
- バックエンド
- インフラ
- セキュリティ
- コスト
- 運用
また、今回も「マルチエージェント構成」が前提となります。
この記事は誰向け？
この記事は以下のような方向けに執筆しています。
- Strands AgentsのGraphを試してみたい方
- Strands Agentsで決定的なマルチエージェント構成を作りたい方
-
workflowとgraphの違いを知りたい方 - Amazon Bedrock AgentCore Runtime上でStrands Agentsを動かしたい方
- AWS CDK（Python）でAgentCore Runtimeへエージェントをデプロイしたい方
デモコードの場所
実際に動かしたい方は、以下を参照ください。
この検証でやりたいこと
今回のゴールは次の3点です。
- Strands Agentsの
GraphBuilderで複数Agentの実行グラフを定義する - 技術選定レビューを複数の専門Agentノードに分割して実行する
- AgentCore RuntimeからGraphを起動し、最終レポートをレスポンスとして返す
構成としては以下のようになります。
正直、これをみただけでは、ワークフローと変わらないです。ここでは構成を理解していただければ良いかなと思います。
そもそも、Graphって何なん？
GraphはStrands Agentsのマルチエージェントパターンの1つです。
GraphBuilderというツールを使って、Agentまたは別のMultiAgentをノードとして追加し、エッジで実行順や依存関係を定義します。
実装のイメージはこんな感じ。
from strands.multiagent import GraphBuilder
builder = GraphBuilder()
builder.add_node(requirements_agent, node_id="requirements_reader")
builder.add_node(frontend_agent, node_id="frontend_reviewer")
builder.add_edge("requirements_reader", "frontend_reviewer")
builder.set_entry_point("requirements_reader")
graph = builder.build()
result = graph("評価対象の技術選定案")
Graphは、前段ノードの出力を後段ノードの入力に自動で含めてくれるので、ワークフローの時みたいなdependenciesの指定がいらないのは開発者にとってはありがたい点かなと思います。
ここで理解しておきたいのは、WorkflowとGraphの比較です。
どちらも「複数の専門Agentを組み合わせて、1つの成果物を作る」ための仕組みですが、設計の主語が少し違います。
| 観点 | Workflow | Graph |
|---|---|---|
| 設計の主語 | タスク | ノード・エッジ |
| 実装の中心 | strands_tools.workflow |
strands.multiagent.GraphBuilder |
| 構成の表現 | タスクの一覧にdescription, system_prompt, dependencies, priorityを定義する |
AgentやMultiAgentをノードとして追加し、エッジで依存関係と情報の流れを定義する |
| 実行順の表現 | タスクごとのdependenciesで表現する |
ノード間のエッジで表現する |
| 状態管理 | workflow_idを持ち、create / start / status / pause / resumeのような操作ができる | graphを構築して呼び出す。実行結果はGraphResultとして取得する |
| 向いているケース | 手順が明確な業務プロセス、進捗確認、再開、タスク単位の管理を重視する処理 | エージェント間の依存関係、分岐、合流、再利用可能な実行グラフをコードで明示したい処理 |
| 拡張性 | タスク定義を増やすことで処理を拡張しやすい | 条件エッジ、サイクル、ネストグラフ、カスタムノードなど構造面の表現力が高い |
| 今回の題材での見え方 | 「要件整理タスク」「レビュータスク」「最終レポートタスク」を順番に処理する | 「要件整理Agent」「各レビューAgent」「判断Agent」「レポートAgent」をnodeとして接続する |
自分は、「タスク同士を繋げることに重点を置いている」のがWorkflow、「エージェント同士を繋げることに重点を置いている」のがGraphかなと理解しました。
実装の流れ
0. 作業プロジェクトの用意
mkdir strands-agents-graph
cd strands-agents-graph
1. CDKプロジェクトの雛形を整える
1-1. 依存パッケージ
CDK側の依存関係は以下です。
aws-cdk-lib==2.255.0
constructs>=10.0.0,<11.0.0
cdk-ecr-deployment>=4.0.0,<5.0.0
開発用にはpytestも入れます。
-r requirements.txt
pytest>=8.0.0,<9.0.0
cdk-ecr-deploymentは、CDKがビルドしたDocker image assetを、Runtime用に作成した専用ECR Repositoryへコピーするために使っています。
Runtimeコンテナ側の依存関係は以下です。
bedrock-agentcore>=1.3.0,<2.0.0
strands-agents>=1.0.0,<2.0.0
boto3>=1.42.0
Graphはstrands-agents本体に含まれているため、workflow版のようにstrands-agents-toolsは不要です。
1-2. CDKエントリーポイント
app.pyは通常のCDKエントリーポイントです。
#!/usr/bin/env python3
import os
import aws_cdk as cdk
from infra.agentcore_tech_selection_graph_stack import AgentCoreTechSelectionGraphStack
app = cdk.App()
AgentCoreTechSelectionGraphStack(
app,
"AgentCoreTechSelectionGraphStack",
env=cdk.Environment(
account=os.environ.get("CDK_DEFAULT_ACCOUNT"),
region=os.environ.get("CDK_DEFAULT_REGION", "ap-northeast-1"),
),
)
app.synth()
1-3. cdk.json
cdk.jsonは、CDK CLIが常にプロジェクト内の仮想環境を使うようにしています。
{
"app": ".venv/bin/python app.py",
"context": {
"@aws-cdk/core:checkSecretUsage": true,
"@aws-cdk/core:target-partitions": [
"aws"
]
}
}
.venv/bin/pythonを固定しておくと、仮想環境をactivateしていない状態でもcdk synthやcdk deployが同じPython環境を使ってくれます。
1-4. 仮想環境
python3.12 -m venv .venv
source .venv/bin/activate
pip install -r requirements-dev.txt -r agent/requirements.txt
AWS認証とリージョンを確認します。
aws sts get-caller-identity
export AWS_REGION=ap-northeast-1
必要であればbootstrapします。
cdk bootstrap
2. Graphのノードを設計する
今回のエージェント実装はagent/tech_selection_graph_agent/に置きます。
agent/
└── tech_selection_graph_agent/
├── agent.py
├── graph_spec.py
└── runtime_app.py
2-1. ノード一覧
ノードIDはgraph_spec.pyに定義します。
前述したように、エージェントの構成は「要件整理エージェント」がまず動いて、その後に並列レビューのエージェントたちが動くような形です。並列レビューのエージェントたちが完了したら、採択エージェントが動き、最終レポートが生成される流れです。
以下のように定数GRAPH_NODE_IDSに、エージェントフローに沿ってノードIDを定義しておきましょう。並列で動作するエージェントは可変長変数を使用しておきましょう。（スプラット演算子を定数（変数）名の前に置くことで可変長定数（変数）になります。）
Pythonだと可変長定数って言わない...？スプラット演算子とか言わない...？Rubyだけ？
GRAPH_NODE_IDS = [
"requirements_reader",
*REVIEW_NODE_IDS,
"decision_maker",
"final_report",
]
次に、並列で動くレビューエージェントたちのノードIDを定義します。上記の可変長変数にノードIDを格納すればOKです。
REVIEW_NODE_IDS = [
"frontend_reviewer",
"backend_reviewer",
"infra_reviewer",
"security_reviewer",
"cost_reviewer",
"operability_reviewer",
]
自分は可変長定数（変数）に並列エージェントを格納して、分けて定義しましたが、GRAPH_NODE_IDSにまとめて定義することもできます。
また、並列で動作するエージェントたちの中で、さらに並列で動くエージェントを定義することもできます。
↓ざっくりとこんな感じで、定義できます。↓
SUB_SUB_PARA_AGENTS = [
"sub_sub_sub_1",
"sub_sub_sub_2"
]
SUB_PARA_AGENTS = [
"sub_sub_1",
*SUB_SUB_PARA_AGENTS,
"sub_sub_3"
]
PARA_AGENTS = [
"sub_1",
"sub_2",
*SUB_PARA_AGENTS,
"sub_4"
]
2-2. 各ノードのsystem prompt
各ノードは独立したAgentなので、それぞれ別のsystem promptを持たせます。
def build_system_prompt(role: str, instruction: str) -> str:
return (
f"あなたは{role}です。\n"
"入力にはOriginal TaskとInputs from previous nodesが含まれることがあります。\n"
"前段ノードの出力を根拠として使い、日本語で具体的に回答してください。\n"
f"{instruction}"
)
たとえば、フロントエンドレビュー用のAgentは以下のように定義しています。
NODE_SYSTEM_PROMPTS = {
"frontend_reviewer": build_system_prompt(
"B2B SaaSのフロントエンドアーキテクト",
"UI開発生産性、型安全性、認証連携、SSR/CSR、パフォーマンス、テスト、チーム運用を評価してください。",
),
}
workflowツールでは、タスク辞書のsystem_promptに役割を書いていました。
Graph版では、各ノードに渡すAgentのsystem promptとして役割を持たせています。
3. Graphを作る
Graph本体はagent/tech_selection_graph_agent/agent.pyに実装します。
3-1. ノード用Agentを作る
各ノードには独立したAgentインスタンスを渡します。
def create_node_agent(node_id: str) -> Agent:
return Agent(
model=create_model(),
system_prompt=NODE_SYSTEM_PROMPTS[node_id],
callback_handler=None,
)
ここで1つ注意点があります。
Graphは、同じAgentオブジェクトを複数ノードに使い回すことを許可していません。
そのため、以下のようにノードごとにAgentインスタンスを作っています。
for node_id in GRAPH_NODE_IDS:
builder.add_node(create_node_agent(node_id), node_id=node_id)
3-2. GraphBuilderでノードとエッジを追加する
def build_graph():
builder = GraphBuilder()
for node_id in GRAPH_NODE_IDS:
builder.add_node(create_node_agent(node_id), node_id=node_id)
builder.set_entry_point("requirements_reader")
for review_node_id in REVIEW_NODE_IDS:
builder.add_edge("requirements_reader", review_node_id)
builder.add_edge(
review_node_id,
"decision_maker",
condition=all_review_nodes_completed,
)
builder.add_edge(
"decision_maker",
"final_report",
condition=decision_completed,
)
builder.set_graph_id("tech_selection_graph")
builder.set_execution_timeout(900)
builder.set_node_timeout(300)
return builder.build()
requirements_readerをエントリーポイントに指定しています。
その後、requirements_readerから各レビュアーにエッジを張ります。
builder.add_edge("requirements_reader", "frontend_reviewer")
builder.add_edge("requirements_reader", "backend_reviewer")
これにより、requirements_readerの完了後に複数レビュアーが実行可能になります。
3-3. 集約ノードには条件を付ける
今回の実装で一番気をつけたのがここです。
GraphのPython実装では、複数入力ノードは「入力エッジのどれか1つがreadyになったら実行可能」と判定されます。
つまり、何も考えずに以下のように書くと、frontend_reviewerだけが終わった時点でdecision_makerが動き始める可能性があります。
builder.add_edge("frontend_reviewer", "decision_maker")
builder.add_edge("backend_reviewer", "decision_maker")
builder.add_edge("infra_reviewer", "decision_maker")
今回は全レビュー結果を見てから意思決定したいので、decision_makerが動き出すためのエッジ条件を付けています。
def all_review_nodes_completed(state: Any) -> bool:
completed_node_ids = {node.node_id for node in state.completed_nodes}
# 全てのレビューエージェントの処理が完了したらTrueを返す
return set(REVIEW_NODE_IDS).issubset(completed_node_ids)
各レビュアーからdecision_makerへのエッジに、同じ条件を付けます。
builder.add_edge(
review_node_id,
"decision_maker",
condition=all_review_nodes_completed, # 上記の条件関数でTrueが返ってきた場合に処理を開始
)
これで、6つのレビュアーがすべて完了してからdecision_makerが実行されます。
final_reportも同様に、decision_maker完了後にだけ実行されるようにしています。
def decision_completed(state: Any) -> bool:
completed_node_ids = {node.node_id for node in state.completed_nodes}
return "decision_maker" in completed_node_ids
4. Graphを実行する
Graphは同期的に呼び出せます。
def execute_tech_selection_graph(prompt: str) -> dict[str, Any]:
graph = build_graph()
result = graph(prompt)
final_report = extract_node_text(result, "final_report")
if not final_report:
final_report = extract_graph_text(result)
return {
"response": final_report,
"graph_status": result.status.value if isinstance(result.status, Status) else str(result.status),
"total_nodes": result.total_nodes,
"completed_nodes": result.completed_nodes,
"failed_nodes": result.failed_nodes,
"execution_order": [node.node_id for node in result.execution_order],
}
result.resultsには各ノードの結果が入ります。
今回欲しいのは最後のfinal_reportなので、そこだけを取り出してRuntimeのresponseにします。
final_report = extract_node_text(result, "final_report")
失敗時やfinal_reportが取れなかった場合に備えて、実行済みノードの結果をまとめるフォールバックも用意しています。
if not final_report:
final_report = extract_graph_text(result)
5. AgentCore Runtimeのentrypointを作る
AgentCore Runtimeから呼ばれるentrypointはagent/tech_selection_graph_agent/runtime_app.pyに置きます。
from bedrock_agentcore.runtime import BedrockAgentCoreApp, PingStatus
from .agent import execute_tech_selection_graph
app = BedrockAgentCoreApp()
入力は{"prompt": "..."}に寄せています。
def invoke_agent(payload: dict) -> dict:
prompt = payload.get("prompt")
if not isinstance(prompt, str) or not prompt.strip():
raise ValueError("payload.prompt must be a non-empty string")
request_id = str(uuid.uuid4())
result = execute_tech_selection_graph(prompt.strip())
return {
"response": result["response"],
"evidence": {
"requestId": request_id,
"graph": {
"status": result["graph_status"],
"totalNodes": result["total_nodes"],
"completedNodes": result["completed_nodes"],
"failedNodes": result["failed_nodes"],
"executionOrder": result["execution_order"],
},
},
}
レスポンスには、最終レポート本文と、Graphの実行情報を含めています。
応答イメージは以下のようになります。
{
"response": "# 技術選定レビュー\n\n## 結論\n条件付き採用...",
"evidence": {
"requestId": "5f3d...",
"graph": {
"status": "completed",
"totalNodes": 9,
"completedNodes": 9,
"failedNodes": 0,
"executionOrder": [
"requirements_reader",
"frontend_reviewer",
"backend_reviewer",
"infra_reviewer",
"security_reviewer",
"cost_reviewer",
"operability_reviewer",
"decision_maker",
"final_report"
]
}
}
}
AgentCore Runtimeのentrypointとhealth checkは以下です。
@app.entrypoint
def invoke(payload: dict) -> dict:
return invoke_agent(payload)
@app.ping
def ping() -> PingStatus:
return PingStatus.HEALTHY
6. Runtimeコンテナを作る
AgentCore Runtimeへ載せるため、エージェントをDocker化します。
FROM public.ecr.aws/docker/library/python:3.12-slim
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
ENV PORT=8080
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY tech_selection_graph_agent ./tech_selection_graph_agent
EXPOSE 8080
CMD ["python", "-m", "tech_selection_graph_agent.runtime_app"]
Graph版では、workflow版のようなSTRANDS_WORKFLOW_DIRは指定していません。
workflowは状態ファイルを保存しますが、Graphは今回の構成ではファイル永続化を使っていないためです。
7. CDKでAgentCore Runtimeを作る
CDKスタックはinfra/agentcore_tech_selection_graph_stack.pyに実装しました。
今回作るリソースは主に以下です。
- ECR Repository
- DockerImageAsset
- ECRDeployment
- AgentCore Runtime
- Runtime用IAM Role
7-1. 専用ECR Repository
image_repository = ecr.Repository(
self,
"AgentImageRepository",
repository_name="agentcore-tech-selection-graph-demo",
image_scan_on_push=True,
image_tag_mutability=ecr.TagMutability.IMMUTABLE,
lifecycle_rules=[
ecr.LifecycleRule(
description="Keep the ten most recent agent images",
max_image_count=10,
)
],
removal_policy=RemovalPolicy.DESTROY,
empty_on_delete=True,
)
検証用途なので、cdk destroyで削除できるようにしています。
7-2. Docker image assetを専用ECRへコピーする
CDKのDocker image assetはCDK管理のasset repositoryにpushされます。
今回はAgentCore Runtime用に専用ECR Repositoryを作っているため、cdk-ecr-deploymentでコピーします。
image_asset = ecr_assets.DockerImageAsset(
self,
"AgentImageAsset",
directory=str(Path(__file__).resolve().parents[1] / "agent"),
platform=ecr_assets.Platform.LINUX_ARM64,
)
image_tag = image_asset.asset_hash
runtime_image_uri = image_repository.repository_uri_for_tag(image_tag)
image_deployment = ecr_deployment.ECRDeployment(
self,
"DeployAgentImage",
src=ecr_deployment.DockerImageName(image_asset.image_uri),
dest=ecr_deployment.DockerImageName(runtime_image_uri),
)
AgentCore RuntimeのコンテナはARM64でビルドしています。
7-3. Runtime用IAM Role
Runtime用ロールには、以下の権限を付与します。
- ECR pull
- ECR authorization token
- CloudWatch Logs
- Bedrock model invoke
runtime_role.add_to_policy(
iam.PolicyStatement(
actions=[
"bedrock:InvokeModel",
"bedrock:InvokeModelWithResponseStream",
],
resources=self._model_resource_arns(model_id),
)
)
model_idがARNで渡された場合はそのARNを使い、それ以外はfoundation modelとinference profileのARNを許可する形にしています。
def _model_resource_arns(self, model_id: str) -> list[str]:
if model_id.startswith("arn:"):
return [model_id]
return [
f"arn:{Aws.PARTITION}:bedrock:*::foundation-model/*",
(
f"arn:{Aws.PARTITION}:bedrock:{Aws.REGION}:{Aws.ACCOUNT_ID}:"
"inference-profile/*"
),
]
7-4. AgentCore Runtime
RuntimeはAWS::BedrockAgentCore::Runtimeを作成します。
runtime = bedrock_agentcore.CfnRuntime(
self,
"AgentRuntime",
agent_runtime_name="AgentCoreTechSelectionGraphRuntime",
description="Strands graph agent that reviews technology selection proposals.",
agent_runtime_artifact=bedrock_agentcore.CfnRuntime.AgentRuntimeArtifactProperty(
container_configuration=bedrock_agentcore.CfnRuntime.ContainerConfigurationProperty(
container_uri=runtime_image_uri,
)
),
network_configuration=bedrock_agentcore.CfnRuntime.NetworkConfigurationProperty(
network_mode="PUBLIC",
),
protocol_configuration="HTTP",
role_arn=runtime_role.role_arn,
environment_variables={
"AWS_REGION": self.region,
"BEDROCK_MODEL_ID": model_id,
},
lifecycle_configuration=bedrock_agentcore.CfnRuntime.LifecycleConfigurationProperty(
idle_runtime_session_timeout=900,
max_lifetime=3600,
),
)
Graph版では、Runtime環境変数はAWS_REGIONとBEDROCK_MODEL_IDだけです。
8. テストを書く
今回も最低限のユニットテストを書いています。
tests/
├── test_agent.py
├── test_agentcore_tech_selection_graph_stack.py
├── test_graph_spec.py
└── test_runtime_app.py
test_graph_spec.pyでは、ノード構成と条件関数を確認しています。
def test_all_review_nodes_completed_requires_every_review_node() -> None:
assert all_review_nodes_completed(state_with_completed(*REVIEW_NODE_IDS))
assert not all_review_nodes_completed(state_with_completed("frontend_reviewer"))
test_runtime_app.pyでは、AgentCore Runtimeのentrypointが期待したレスポンス形式を返すか確認しています。
CDKのテストでは、Runtime、ECR Repository、IAM Policy、Outputが作られることを確認しています。
手元では以下でテストしました。
env JSII_RUNTIME_PACKAGE_CACHE_ROOT=/tmp/jsii-package-cache \
JSII_SILENCE_WARNING_UNTESTED_NODE_VERSION=1 \
.venv/bin/pytest
結果は以下です。
11 passed
9. デプロイして確認する
デプロイは通常どおりです。
pytest
cdk synth
cdk deploy
別のBedrockモデルIDを使う場合は、CDK contextで指定できます。
cdk deploy -c modelId=your-model-or-inference-profile-id
デプロイできたら、AWSマネジメントコンソールにアクセスして、Amazon Bedrock AgentCore Runtimeのコンソールへ遷移します。
そうすると、デプロイしたランタイムが表示されているかと思いますので、クリックして、画面右上の「テスト」をクリックします。
そうしたら、「入力」のエディタにJSON形式で入力文を入れます。
{
"prompt": "新規B2B SaaSを、Next.js + FastAPI + Aurora PostgreSQL + ECS Fargate + Bedrockで作る案を評価して"
}
入力したら、「実行」をクリックしましょう。
複数のサブエージェントによるタスクを実行しているので、レスポンスにはそこそこ時間がかかります。
処理が完了すると、出力のresponseに技術選定レポートが入っていることが確認できるかと思います。
evidence.workflow に workflow の実行情報が入ります。
{
"response": "# 技術選定レポート：新規B2B SaaS スタック評価\n## Next.js + FastAPI + Aurora PostgreSQL + ECS Fargate + Bedrock\n\n**作成者：** シニアソフトウェアアーキテクト \n**対象読者：** 経営・開発リード \n**判定日：** 2025年 \n**ステータス：** 🟡 **条件付き採用**\n\n---\n\n## 結論\n\n本スタックは **技術的合理性が高く、B2B SaaSの要件に適合している**。AWSネイティブで統一されたエコシステム、Python中心の一貫した開発体験、Bedrockによる顧客データ保護の担保は、競合スタックと比較して明確な優位性を持つ。\n\nただし、**現時点では「セキュリティ設計の空白」が多く、このまま機能開発を開始するのは危険**である。マルチテナント分離・認証基盤・コスト制御の3点が未設計のまま本番を迎えた場合、B2B契約上の致命的インシデントにつながるリスクがある。\n\n**P0条件（後述）を Sprint 1 内に完了した時点で「正式採用」に格上げする。**\n\n---\n\n## 採用条件\n\n採用を正式確定するために、以下を**優先度順・期限付き**で完了すること。\n\n### 🔴 P0｜MVP開発着手前に必須完了（Week 1〜2）\n\n| # | 項目 | 具体的要件 | 理由 |\n|---|------|-----------|------|\n| P0-1 | **マルチテナント分離の実装** | Aurora に Row Level Security (RLS) を設定 + FastAPI BaseRepository に `tenant_id` 強制フィルタ + Integration Test でテナント間アクセス不能を証明 | 後付け修正のコストが全項目で最大。設計ミスはB2B致命的インシデントに直結 |\n| P0-2 | **認証基盤の確立** | Amazon Cognito（SAML/SSO対応）を採用し、JWT検証 Depends を全エンドポイントに適用。RBAC（OWNER/ADMIN/MEMBER/VIEWER）モデル確定 | 認証は後付け不可。全機能の設計前提になる |\n| P0-3 | **シークレット管理の徹底** | DB接続文字列・JWT秘密鍵を全件 AWS Secrets Manager に移行。ECS環境変数への直書き禁止を pre-commit hook で強制 | `.env` のGitコミット1件で全テナントデータが危険に晒される |\n| P0-4 | **IAM 最小権限設計** | ECSタスクロールと実行ロールを分離。Bedrock IAMポリシーは使用モデルARNのみに限定 | 後から修正すると本番サービスへの影響が発生する |\n\n> **P0が完了した時点で本レポートの判定を「採用」に更新する。**\n\n---\n\n### 🟠 P1｜Sprint 1〜2 で完了（Week 3〜4）\n\n| # | 項目 | 具体的要件 |\n|---|------|-----------|\n| P1-1 | **Bedrockコスト制御の実装** | テナント別トークン使用量をAuroraに記録 + プラン別月次上限 + CloudWatch Billing Alert（$200/$500/$1,000の3段階）+ 軽量タスク→Haiku・複雑処理→Sonnet のモデルTier分岐 |\n| P1-2 | **boto3非同期設計の標準化** | `run_in_executor` パターンをADRに明文化 + `LLMProvider` Protocol インターフェース定義（`BedrockProvider` / `MockLLMProvider`）+ CIでイベントループブロッキングをテスト |\n| P1-3 | **RDS Proxy の初期導入** | Fargate → RDS Proxy → Aurora の接続経路を構築。Auroraフェイルオーバー時の接続維持に必須。後付けすると接続文字列変更でアプリ修正が発生 |\n| P1-4 | **DBマイグレーション手順の整備** | Expand-Contract パターンをチームルール化 + Alembic実行をアプリデプロイと独立したCIジョブに分離 |\n| P1-5 | **Bedrock VPC エンドポイント設定** | `bedrock-runtime` の VPC Interface Endpoint を構成し、Bedrock通信がインターネットに出ないことを確認（NAT Gatewayコスト削減 + データ保護の両立） |\n\n---\n\n### 🟡 P2｜MVPリリース後3ヶ月以内（運用成熟化）\n\n- CloudTrail + Bedrock Model Invocation Logging の有効化\n- 構造化ログ（structlog）+ CloudWatch ダッシュボード\n- AWS X-Ray による分散トレーシング\n- OpenAPI → TypeScript型自動生成パイプラインのCI組み込み\n- Bedrock Guardrails（プロンプトインジェクション対策）\n- 月次バックアップリストアテストの定期実施\n- SLO/SLI の定義（推奨：可用性99.5%から開始）\n\n---\n\n## 主要リスク\n\n| リスク | 発生確率 | 影響度 | 対策（優先度） |\n|--------|---------|--------|---------------|\n| **テナント間データ漏洩** | 中 | 🔴 最高 | RLS + Repository 二重チェック + Integration Test（P0-1） |\n| **Bedrockコスト急増** | 高 | 🟠 高 | テナント別上限 + モデルTier分岐 + 日次アラート（P1-1）|\n| **boto3 async 設計ミス**（イベントループブロッキング） | 高 | 🟡 中 | ADR明文化 + CIテスト（P1-2） |\n| **DBマイグレーション失敗** | 低 | 🟠 高 | Expand-Contract + 手順書 + CI分離（P1-4） |\n| **プロンプトインジェクション** | 中 | 🟡 中〜高 | システムプロンプト分離設計 + Guardrails（P2） |\n| **Aurora接続枯渇** | 中 | 🟡 中 | RDS Proxy（P1-3）+ `pool_pre_ping=True` |\n\n### ⚠️ 経営レベルで注視すべきリスク：Bedrockコスト構造\n\n成長期（契約50社）以降、**Bedrockがインフラコスト全体の54%を占める**構造になる。SaaS料金設計をBedrockのトークン消費に連動させなければ、スケールするほど損益が悪化するビジネスリスクがある。\n\n```\n【フェーズ別インフラコスト概算】\n\n MVP期 （5社） ：月 約 $180 ≒ 約 27,000円\n 成長期 （50社） ：月 約 $1,400 ≒ 約 210,000円\n スケール（300社）：月 約 $7,800 ≒ 約 1,170,000円\n\n【料金設計の最低ライン（インフラコスト比率 < 10% を維持する場合）】\n ライトプラン （50万トークン/月） → 最低 $50/社/月\n スタンダード （300万トークン/月）→ 最低 $150/社/月\n```\n\n---\n\n## 代替案\n\n本スタックを採用しない場合、または一部を置き換える場合の選択肢を示す。\n\n### バックエンド代替：Node.js（Express / Hono）\n\n| 観点 | 評価 |\n|------|------|\n| 採用理由 | フロントエンドチームがTypeScript統一を望む場合 |\n| デメリット | BedrockおよびAWS SDK統合でPythonほどの成熟度がない。boto3エコシステムを失う |\n| **推奨度** | ❌ AI機能が主要価値の場合はFastAPIを維持すべき |\n\n### LLM基盤代替：Azure OpenAI / OpenAI API\n\n| 観点 | 評価 |\n|------|------|\n| 採用理由 | GPT-4oの能力を直接利用したい場合 |\n| デメリット | 顧客データがAWS外に送信される。B2B契約でのデータ保護説明責任が困難になる。ISMS/SOC2取得の障壁になりうる |\n| **推奨度** | ⚠️ B2Bデータ保護要件がない場合は検討余地あり。エンタープライズ顧客を取る場合はBedrockを維持 |\n\n### インフラ代替：EKS（Kubernetes）\n\n| 観点 | 評価 |\n|------|------|\n| 採用理由 | マイクロサービス化が確定している場合、将来のマルチクラウド対応 |\n| デメリット | 初期運用コスト・学習コストがECS Fargateの3〜5倍。MVP段階ではオーバーエンジニアリング |\n| **推奨度** | ❌ 300社超え・月$10,000以上のインフラ規模になってから再評価 |\n\n### 認証代替：Auth0 / Clerk\n\n| 観点 | 評価 |\n|------|------|\n| 採用理由 | 開発速度優先。SSOとSAML設定のUI/UXがCognitoより優れる |\n| デメリット | AWSエコシステムから外れるためVPC連携が複雑化。有料プランのコスト増 |\n| **推奨度** | 🟡 開発初期スピードを最優先する場合は検討余地あり。本番移行前にCognito移行を計画すること |\n\n---\n\n## 次の検証タスク\n\n開発着手前・着手後に確認すべき検証タスクを具体的に列挙する。\n\n### 着手前（今週中）\n\n```\n□ 1. マルチテナント設計の事前レビュー\n └ Aurora RLS ポリシーのPoCをローカル環境で実装し、\n テナント分離がSQLレベルで機能することを確認する\n\n□ 2. Bedrock利用可能リージョン・モデルの確認\n └ Anthropic Claude 3.x / Titan Embeddings が\n ターゲットリージョン（ap-northeast-1）で利用可能かを確認\n ※ 一部モデルはリージョン制限あり\n\n□ 3. チームのNext.js App Router習熟度調査\n └ RSC / Server Actions の実装経験者が何名いるかを確認\n ゼロの場合はSprint 0にハンズオン研修を設ける\n\n□ 4. SaaS料金設計の事前合意\n └ Bedrockコストがビジネスモデルを破壊しないよう、\n トークン消費量と料金プランの対応表を経営と合意する\n```\n\n### Sprint 1 完了時（2週間後）\n\n```\n□ 5. P0条件の達成確認レビュー\n └ 以下4点のコードレビュー + 動作確認を実施\n ・RLS + Repository二重チェック + Integration Test PASS\n ・Cognito認証 + JWT検証の全エンドポイント適用\n ・Secrets Manager移行 + .env Git禁止の確認\n ・IAMロール分離・最小権限の設定確認\n\n□ 6. コスト実測値の確認\n └ 開発環境1週間分のAWSコストを確認し、\n $180/月（MVP期）の試算と乖離がないことを検証\n```\n\n### Sprint 2 完了時（4週間後）\n\n```\n□ 7. Bedrockコスト制御の動作確認\n └ テナント別トークンカウントが正しく記録・集計されることをテスト\n 上限到達時に429が正しく返却されることを確認\n\n□ 8. 負荷テスト（小規模）\n └ k6で10並列・1分間のリクエストを送り、\n 接続プール枯渇・イベントループブロッキングが発生しないことを確認\n\n□ 9. セキュリティスキャン\n └ Trivy（コンテナイメージ）+ Bandit（Pythonコード）を\n CIパイプラインに組み込み、CRITICALゼロを確認\n```\n\n---\n\n## まとめ：意思決定のポイント\n\n| 問い | 答え |\n|------|------|\n| このスタックで作れるか？ | ✅ **Yes**。技術的合理性は高い |\n| 今すぐ機能開発を始めてよいか？ | ❌ **No**。P0の4条件を先に完了すること |\n| Bedrockは使うべきか？ | ✅ **Yes**。ただしコスト制御機構の実装を料金設計と同時に行うこと |\n| いつ「正式採用」になるか？ | **Sprint 1 完了レビューでP0条件を充足した時点** |\n\n> 「**作れる技術スタック**」から「**安全に運用できる技術スタック**」に昇格させるための投資がP0条件である。この2週間の設計投資を惜しんだ場合、後から支払うコストは10〜100倍になる。\n\n---\n\n*本レポートはdecision_makerノードの統合評価結果を根拠として作成。次回更新：Sprint 1完了レビュー後*",
"evidence": {
"requestId": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
"graph": {
"status": "completed",
"totalNodes": 9,
"completedNodes": 9,
"failedNodes": 0,
"executionOrder": [
"requirements_reader",
"cost_reviewer",
"frontend_reviewer",
"backend_reviewer",
"infra_reviewer",
"operability_reviewer",
"security_reviewer",
"decision_maker",
"final_report"
]
}
}
}
出力のevidenceは、自分が任意で出力しているものです！
Graphを使ってみた所感
まず、比較対象となるworkflowは「タスク定義を辞書型で定義して、サブエージェントに実行してもらう」感じでした。主語が「ワークフロー内のタスク」でした。
一方で、Graphは「各エージェントをつなげて、フローを作る」感じです。
正直な感触だと、WorkflowとGraphのどちらが優れているか優劣がつけづらいなぁ...って感じです。
処理すべきタスクと順番が決まっていて、タスクを繋げてエージェントフローを構築したいという場合には、Workflowを使うのが良いかもしれません。また、各タスクの状態管理ができるのがWorkflowの強みでもあるので、状態管理をしっかりしたいケースには向いていますね。
Graphを使いやすいケースは、エージェントの出力によって、次に実行する処理を分岐させたい場合だと思います。Graphでは、conditionの引数で条件関数を指定できます。これを使用することで、次に実行する処理を分岐させることができます。
まだ試せていないので憶測になってしまいますが、GraphとStructuredOutputは相性が良いかもしれません。
構築をしていた際の注意点は以下です。
- 各ノードには別々のAgentインスタンスが必要
- 複数入力ノードはデフォルトだとOR的にreadyになり得るため、集約ノードには条件関数を付けた方がよい
- Graphは状態ファイルを自動保存するツールではないため、実行履歴を永続化したい場合は別途設計が必要
- ノード数が増えると、エッジ定義をどう整理するかが重要になる
まとめ
今回は、Strands AgentsのGraphBuilderを使って、マルチエージェント構成をAgentCore Runtime上にデプロイする構成を作りました。
やったことは以下です。
- 技術選定レビューを9つのAgentノードに分割
-
GraphBuilderでノードとエッジを定義 - 全レビュー完了後に
decision_makerが動くよう条件関数を追加 -
final_reportノードの結果をRuntimeレスポンスへ返却 - CDKでECR、AgentCore Runtime、IAM Roleを作成
WorkflowとGraphの使い分けはズバリ、「成果物を生成するためのタスクと順番が決まっている場合はWorkflow」、「出力された内容によって次に実行するタスクを決めたい場合にはGraph」というような決め方をするのが良いかもしれないですね。
また、処理に時間はかかってしまいますが、大枠のマルチエージェント構成はGraphで作成し、各タスクの実行をWorkflowツールを使って詳細に定義をするのが、堅牢なマルチエージェント構成になるのではないかなと思っています。
これ、Claude Codeで手元でSkill化したい...
以上