7月29日、AWSが「Market surveillance agent with LangGraph and Strands on AgentCore」と題した記事を公開した。LangGraphとStrands、Amazon Bedrock AgentCoreを組み合わせた本番運用可能なマルチエージェントシステムの設計・デプロイ手法を詳しく解説した内容だ。特筆すべきは、LangGraphがマクロ制御を、StrandsがノードレベルのLLM推論を担うという明確な役割分担と、LLMに直接SQLを生成させないツール設計によるSQLインジェクション対策の具体的な実装例が示されている点である。
なぜLangGraph×Strandsなのか
マルチエージェントシステムの構築において、エンジニアはしばしば「ワークフロー全体をLLMに任せるか、それとも決定論的に制御するか」という二択を迫られる。この記事が提示するアーキテクチャは、その二択を回避する。
LangGraphがマクロ制御を担い、StrandsがノードレベルのLLM推論を担うという役割分担だ。
- LangGraph：有向グラフでワークフローを定義し、チェックポイントによる障害復旧、Human-in-the-loop対応、条件分岐やParallel実行を管理する
- **Strands Agent**：AWSが開発したオープンソースのエージェントSDKで、個々のノード内でモデル非依存の推論ループを実行し、ツール呼び出しと中間結果の評価を繰り返す。Claude・Titan・サードパーティモデルを問わず同一インターフェースで扱える点が特徴だ
記事の言葉を借りれば「決定論的なオーケストレーションの中に、局所的・動的なインテリジェンスを配置する」設計だ。純粋にLLMへ処理を委ねると、本番では非決定的な動作がリスクになる。一方、完全なルールベースでは複雑な分析タスクに対応できない。この二層構造はその折衷点として機能する。
ユースケース：金融市場の監視エージェント
デモとして構築されているのは、金融市場の不審取引を検出・調査・レポート化するマルチエージェントシステムだ。以下の4つの専門エージェントが連携する。
security_monitor：価格・出来高・ティックレベルの取引を分析broker_monitor：ブローカー別の取引パターンを監視risk_monitor：リスク指標を評価intel_analyst：情報を統合して調査判断を行う
オーケストレーターが受け取ったクエリを解析し、必要な専門エージェントのリストと各タスクを決定。route_analysts関数がインデックスをたどりながら順次ノードを呼び出し、最後にsynthesizerが全エージェントの知見をまとめてレポートを生成する。
SQLインジェクション対策としてのツール設計
技術的に面白いのが、Strandsのツール設計におけるセキュリティへの配慮だ。
LLMに直接SQLを生成させると、プロンプトインジェクション経由でSQLインジェクションが発生するリスクがある。これを避けるため、ツールを以下の3層に分離している。
get_report_list：利用可能なレポートの一覧を返すget_report_schema：レポートのカラム定義を返すrun_report：スキーマで許可されたカラムのみをフィルタに使い、バインドパラメータでSQLを構築・実行する
# LLMはSQLを直接書かない。フィルタはスキーマで検証されたカラムのみ
unknown = set(filters) - allowed_columns
if unknown:
raise ValueError(
f"Unknown filter field(s) {sorted(unknown)} for {report_name}. "
f"Allowed: {sorted(allowed_columns)}"
)
where = " AND ".join(f"{field} = :{field}" for field in filters)
sql = f"SELECT * FROM {schema['reportName']}"
LLMはSQLを一切書かない。フィルタ値もバインドパラメータとして渡されるため、クエリ文字列への注入が構造的に不可能になっている。
LangGraphのワークフロー実装
各専門ノードは以下のパターンで実装される。ノードが起動するたびに独立したStrandsエージェントを生成し、オーケストレーターが共有ステートに書き込んだタスクを受け取って実行する。
async def security_monitor_node(state: AgentState) -> AgentState:
agent = Agent(
name="security_monitor",
model=analyst_model,
system_prompt=SECURITY_MONITOR_PROMPT,
tools=[get_report_list, get_report_schema, run_report],
)
task = state.get("agent_task_map", {}).get("security_monitor", state["query_text"])
chunks = []
async for event in agent.stream_async(task):
if "data" in event:
chunks.append(event["data"])
return {
"security_monitor_insights": {"task": task, "business_insights": "".join(chunks)},
"current_agent_index": state.get("current_agent_index", 0) + 1,
}
各ノードのStrandsエージェントはコンテキストを独立して保持し、LangGraphが共有ステートをつなぎ合わせる。エージェントをひとつのモノリスとして構成した場合、会話が長期化するにつれてコンテキストウィンドウが圧迫され、推論精度が低下するリスクがある。本アーキテクチャは各エージェントのコンテキストを分離することで、この問題を構造的に回避している。
チェックポイントにはAgentCoreMemorySaverを使用し、全ノード実行後にステートを自動スナップショットする。障害時はそのチェックポイントから再開できる。
Amazon Bedrock AgentCoreへのデプロイ
**Amazon Bedrock AgentCore**はエージェントの本番運用を管理するフルマネージドサービスで、コンテナのオーケストレーション、スケーリング、セッション管理を自動で処理する。LangGraphとStrandsをそのままサポートしており、AgentCore Python SDKを使ってローカルのエージェントコードをクラウドネイティブにデプロイできる。
ランタイムは用途に応じて2種類が用意されている。拡張ランタイムは長時間実行される調査・分析タスク向けで、タイムアウトを気にせず複数エージェントが連鎖的に処理を行うユースケースに適する。一方、低レイテンシ実行はインタラクティブなワークフローや応答速度が優先される場面を想定しており、Human-in-the-loopを組み込んだ対話型エージェントとの親和性が高い。本記事のマーケットサーベイランスのような非同期バッチ調査には拡張ランタイムが適切な選択となる。
完全なソースコードはGitHubで公開されている。
詳細はMarket surveillance agent with LangGraph and Strands on AgentCoreを参照していただきたい。
