AWS AgentCore、LangGraphとStrandsを活用した市場監視エージェントを公開

AI アプリケーションが単純なチャットボットから洗練された自律システムへと進化する中、組織は現実世界の生産環境に対応できる複雑なマルチエージェントワークフローを調整するという新たな課題に直面しています。従来の単一エージェントのアプローチでは、専門知識の活用や動的な意思決定、堅牢なエラー回復メカニズムが求められる複雑なビジネスプロセスには対応しきれないケースが多々あります。

金融サービス業界はこの課題を象徴する好例です。市場監視システムは、取引パターンの分析、不審活動の調査、包括的なレポート作成といったタスクを遂行しつつ、厳格なコンプライアンスと信頼性の基準を維持するために、複数の専門エージェントを調整する必要があります。

本ソリューションは、2 つのフレームワークを組み合わせています。マクロレベルのワークフローオーケストレーションには LangGraph を、インテリジェントなエージェント推論には Strands を活用します。

LangGraph は、マルチエージェント間の調整において状態管理と有向グラフの処理に優れています。これにより、ワークフローの実行と、エージェント間で共有される状態に対して細かな制御が可能になります。また、中央集約型の永続化層を備えており、人間がループに参加するインタラクションや、障害発生時の堅牢なチェックポイントベースの復旧など、本番環境で重要な機能をサポートしています。

一方、Strands エージェントは、個々のワークフローノード内で推論エンジンとして機能します。モデルに依存しない能力を持ち、さまざまな大規模言語モデル（LLM）プロバイダーと統合しながら、柔軟なツール連携と包括的な観測性を提供します。

昨年の Amazon Bedrock AgentCore のリリースにより、多くのユースケースでエージェント型ソリューションの本番導入が簡素化される可能性があります。この組み合わせは、複雑なユースケースを処理しつつ、エンタープライズアプリケーションが求めるインフラの信頼性と観測性を実現する、本番対応のエージェント AI システムのための堅固な基盤となります。

本記事では、AWS インフラ上で LangGraph と Strands を活用してマルチエージェント AI システムを設計・デプロイする方法を紹介します。LangGraph のチェックポイントシステムを用いた状態駆動型のワークフロー制御や、専門的な推論タスクに特化した Strands エージェントの統合、そしてスケーラブルな本番環境での運用を実現する AgentCore の活用方法について解説します。

完全な実装は GitHub で公開されています。

Strands：インテリジェントなエージェント推論

Strands エージェントは、既存のインフラに適合するモデル非依存アーキテクチャを採用しており、独自の制約を課すことなく柔軟に動作します。このエージェントは、ツールの出力を継続的に評価し、中間結果に基づいて判断を下す「アジェンティック・ループ（推論ループ）」を実装しています。これにより、複雑な多段階分析ワークフローの構築が可能になります。

また、フレームワークには包括的なセッション管理とステート管理機能、そしてコンテキストウィンドウが溢れるのを防ぐための複数の会話マネージャーが含まれています。

Strands を使えば、ツールスキーマやアクセスパターンを定義することで、外部ツールの連携を柔軟に設定できます。本稿の監視エージェントでは、ハルシネーション（幻覚）を防ぎ、インジェクション攻撃への耐性を高めるため、「データの発見」と「データ取得」のプロセスを分離しています。具体的には、get_report_list や get_report_schema といったツールを使ってレポートを検索し、検証済みのパラメータで SQL クエリを構築して実行する run_report ツールを活用します。

以下に、セキュリティ監視用エージェント（security_monitor）のシステムプロンプトと使用ツールの構成を示します。

Strands の Agent と Bedrock モデルを活用した、市場監視エージェントの構築例です。

まず、必要なライブラリをインポートします。strands から Agent クラスと tool デコレータを読み込み、strands.models.bedrock から BedrockModel を読み込みます。

次に、使用するモデルの設定を行います。ここでは Anthropic の Claude Sonnet 4.6 (us.anthropic.claude-sonnet-4-6) を米国東部リージョン（us-east-1）で利用します。最大トークン数は 16,000 に設定し、思考プロセスには適応型のアダプティブ・シンキングを適用して、8,000 トークンの予算を割り当てています。また、プロンプトのキャッシュはデフォルト設定を使用します。

次に、エージェントが利用可能なレポート一覧を取得するためのツール get_report_list を定義します。このツールは特定のエージェント名（例：'security_monitor'）を受け取り、そのエージェントに関連する利用可能なレポートの名前と説明を含む JSON 配列を返します。内部では load_agent_reports 関数を呼び出してデータを取得し、整形された JSON 文字列として出力します。

続いて、レポートのクエリ構築に必要なカラム定義を取得するためのツール get_report_schema を用意します。このツールは、対象となるレポート名（例：'TradeActivity'）と、抽出したいデータの意図（query_intent）を受け取ります。内部で load_json_report_definition 関数を呼び出してパラメータとカラム定義を含む JSON オブジェクトを取得し、整形された文字列として返します。

最後に、事前定義済みレポートを実行するツール run_report を定義します。このツールはレポート名、フィルタ条件（辞書形式）、および結果の上限値を受け取ります。重要なのは、このツールがすべてのフィルタをレポートのスキーマに対して検証し、パラメータ化された SQL クエリを構築することです。これにより、LLM が生 SQL を直接記述する必要がなくなり、SQL インジェクションによるセキュリティリスクを防ぐことができます。

引数:

report_name: get_report_list から取得したレポート名（例：'TradeActivity'）。

filters: カラム名をキーとする等値フィルタ。例：{"symbol": "AAPL", "date": "2024-03-15"}。

limit: 任意の行数上限（1〜10,000）。

戻り値:

dict: {'success': bool, 'data': str (CSV), 'error': str または None}

"""

schema = load_json_report_definition(report_name)

allowed_columns = {c["name"] for c in schema["columns"]}

レポートで許可されているカラムリストに含まれないフィルタフィールドがあれば拒否する。

unknown = set(filters) - allowed_columns

if unknown:

raise ValueError(

f"レポート '{report_name}' で不明なフィルタフィールド {sorted(unknown)} が検出されました。"

f"許可されているカラム: {sorted(allowed_columns)}"

)

名前付きバインドパラメータを使用した SQL を構築する。

where = " AND ".join(f"{field} = :{field}" for field in filters)

sql = f"SELECT * FROM {schema['reportName']}"

if where:

sql += f" WHERE {where}"

if limit is not None:

if not isinstance(limit, int) or not 1

LangGraph: マクロワークフローのオーケストレーション

LangGraph は、複雑な AI ワークフローに適した 3 つのコア機能を通じて、マルチエージェントシステムの生産レベルでのオーケストレーションを提供します。

グラフベースの状態機械: LangGraph は、エージェントのワークフローを有向グラフとしてモデル化します。ここでノードはエージェントロジックを含む関数を表し、エッジが実行フローを決定します。この宣言的なアプローチにより、複雑な多段階推論を読みやすく保守可能なコードに変換できます。グラフは条件分岐、並列実行、動的ルーティングをサポートしており、中間結果に基づいてワークフローが適応する実世界シナリオにおいて重要な機能です。

永続的な状態管理: このフレームワークのチェックポイントシステムは、ノードの実行完了後にワークフローの状態を自動的にスナップショットします。これらのチェックポイントにより、障害からスムーズに回復したり、人間が関与するインタラクションをサポートしたりできます。アナリストが中間結果を確認する場合やエラーが発生した場合でも、システムは正確なチェックポイントから復元され、以前の作業を失うことなく再開されます。この状態管理アーキテクチャは、多段階の会話、反復的な改善、数時間または数日にわたる長期調査など、一般的に使用されるパターンを支えています。

本番環境での信頼性: LangGraph には、スロットリングやシステム内で発生しうる他の障害への対応のための、指数バックオフを備えた組み込みのリトライ戦略が用意されています。また、OpenTelemetry を通じて包括的な観測性を提供するため、主要な観測アプリケーションとの統合も容易です。

専門的な Strands エージェント間でクエリをルーティングするオーケストレーション層を構築しましょう。以下のコードは、共有状態、どの専門エージェントを呼び出すかを選択するオーケストレーター、それらの間での条件付きルーティング、および回復や人間によるレビューを可能にするチェックポイントベースの永続化機能を備えたワークフローグラフを定義しています。

from typing import TypedDict, Optional, List, Dict, Any

from langgraph.graph import END, StateGraph

from langgraph_checkpoint_aws import AgentCoreMemorySaver

class AgentState(TypedDict):

query_text: str

session_id: Optional[str]

agent_task_map: Optional[Dict[str, str]]

required_agents: Optional[List[str]]

current_agent_index: Optional[int]

# 各専門エージェントはここに洞察を記述します

security_monitor_insights: Optional[Dict[str, Any]]

broker_monitor_insights: Optional[Dict[str, Any]]

risk_monitor_insights: Optional[Dict[str, Any]]

intel_analyst_insights: Optional[Dict[str, Any]]

synthesizer_insights: Optional[str]

SPECIALIST_NODES = {

"security_monitor": security_monitor_node,

"broker_monitor": broker_monitor_node,

"risk_monitor": risk_monitor_node,

"intel_analyst": intel_analyst_node,

}

def route_analysts(state: AgentState) -> str:

"""動的ルーティング --- required_agents リストをインデックス順に走査します。"""

required = state.get("required_agents", [])

index = state.get("current_agent_index", 0)

if not required:

return END

if index

市場分析エージェントにおける LangGraph ワークフローのオーケストレーション

なぜ LangGraph と Strands を組み合わせるのか

多くの企業ユースケースでは、厳格で事前に定義されたワークフローが不可欠です。LLM の非確定的な性質に頼って正しい手順を実行させるのはリスクがあります。しかし、そのワークフロー内の特定のステップには、LLM が持つ俊敏な推論能力と知性が求められます。

LangGraph と Strands を組み合わせることで、このギャップを埋めることができます。これにより、決定論的なオーケストレーションが局所的で動的な知能を内包するシステムを構築できます。

このアーキテクチャの組み合わせが、複雑なワークフローの課題をどのように解決するかを見てみましょう。

「ノード」レベルでの知能: LangGraph はワークフロー全体の上位レベルのオーケストレーションを定義します。一方、Strands エージェントは特定のノードに配置されます。複雑なワークフローにおいて LLM による分析や曖昧さへの対応が必要な場合に、このノードが機能します。自律的な推論とツールの使用は、その柔軟性が厳密に必要とされる箇所でのみ適用されます。

ノード型エージェントによるコンテキストの分離: モノリス型のエージェントは、指示を把握し続けることが容易ではありません。個別の Strands エージェントを独立した LangGraph ノード内に配置することで、メモリを区画化できます。各エージェントが自身の超焦点化されたコンテキストとツールの履歴を管理する一方、LangGraph は構造化された包括的なセッション状態を維持します。これにより、異なるエージェントは互いに独立してこの状態を更新したり参照したりすることが可能になります。

オーケストレーションの強化: LangGraph の中核は、低レベルなルーティングとオーケストレーションツールです。ここに Strands を埋め込むことで、包括的なエンタープライズエージェントフレームワークをグラフに即座に組み込めます。LangGraph の堅牢なルーティング機能に加え、Strands 独自のモデルコンテキストプロトコル（MCP）連携、ステアリング制御、セーフティガードレール、評価機能を同時に活用できます。

具体的には、各専門家が LangGraph ノードとして機能します。ノードはシステムプロンプト、ツール、孤立した文脈を持つ新しい Strands エージェントを起動し、オーケストレーターから割り当てられたタスクを実行して状態更新を返します。LangGraph はこの部分的な更新を、他のノードが参照する共有状態に統合します。以下がセキュリティ監視ノードの例です。

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],
        callback_handler=None,
    )

# オーケストレーターが共有状態に設定したタスクをこのノードから取得します。
    task = state.get("agent_task_map", {}).get("security_monitor", state["query_text"])

Strands は独自の推論とツールループを実行し、エージェントの最終的なテキストを収集します。

chunks = []
async for event in agent.stream_async(task):
    if "data" in event:
        chunks.append(event["data"])
result = "".join(chunks)

共有状態の更新を返却します。

return {
    "security_monitor_insights": {"task": task, "business_insights": result},
    "current_agent_index": state.get("current_agent_index", 0) + 1,
}

AgentCore を活用した AWS インフラとデプロイ

Amazon Bedrock AgentCore は、スケーラブルなエージェントの展開と運用を完全に管理するサービスです。インフラ管理の手間を減らしながら、本番環境で使える機能を提供します。

ランタイムでのデプロイ：AgentCore ランタイム（Amazon Bedrock AgentCore の機能）は、最小限の設定でローカルのエージェントコードをクラウドネイティブな展開に変換します。このサービスはフレームワークに依存せず、LangGraph や Strands と相性が良く、すぐに利用可能です。動的なエージェントワークロード向けに特別に設計されたインフラを提供し、長時間続く調査のための拡張ランタイムや、対話型ワークフロー用の低遅延実行、需要に応じた自動スケーリングが含まれています。

Amazon Bedrock AgentCore の機能である Python SDK とスターターツールキットを活用すれば、LangGraph オーケストレーターと Strands エージェントを簡単にデプロイできます。ランタイムがコンテナ化、ネットワーク設定、計算リソースの割り当てを自動で処理します。AgentCore は、コンテナのオーケストレーション、スケーリング、セッション管理といった本質的ではない重労働を一手に引き受けます。

api.py --- AgentCore ランタイムのエントリーポイント

from bedrock_agentcore.runtime import BedrockAgentCoreApp

from src.agents import Workflow

app = BedrockAgentCoreApp()

workflow = Workflow()

@app.entrypoint

async def market_surveillance_workflow(payload):

"""各リクエストに対して AgentCore から呼び出されます。ストリーミング形式でチャンクを返します。"""

prompt = payload.get("prompt")

session_id = payload.get("session_id", "default-session")

actor_id = payload.get("actor_id", "default-actor")

async for chunk in workflow.stream_query(

session_id=session_id, prompt=prompt, actor_id=actor_id

):

yield chunk

if __name__ == "__main__":

app.run()

AgentCore スターターツールキットでデプロイする

from bedrock_agentcore_starter_toolkit import Runtime

runtime = Runtime()

runtime.configure(

entrypoint="api.py",

auto_create_execution_role=True,

auto_create_ecr=True,

requirements_file="requirements.txt",

region="us-east-1",

agent_name="market_surveillance_workflow",

)

result = runtime.launch()

print(f"Agent ARN: {result.agent_arn}")

デプロイしたエージェントを呼び出す

import boto3, json

client = boto3.client("bedrock-agentcore", region_name="us-east-1")

response = client.invoke_agent_runtime(

agentRuntimeArn=result.agent_arn,

qualifier="DEFAULT",

payload=json.dumps({

"prompt": "What caused the AAPL price spike at 11:00 AM on March 15, 2024?",

"session_id": "session-001",

"actor_id": "analyst-jane",

}),

)

AgentCore はサーバー送信イベント（SSE）ストリームを返します。これを解析します。

for raw in response["response"].iter_lines():

if not raw:

continue

line = raw.decode("utf-8") if isinstance(raw, bytes) else raw

if not line.startswith("data: "):

continue

try:

chunk = json.loads(line[6:])

if isinstance(chunk, str) and chunk.startswith("data: "):

chunk = json.loads(chunk[6:])

except json.JSONDecodeError:

continue # データ形式が崩れているチャンクは無視して、クラッシュは回避

if isinstance(chunk, dict) and chunk.get("type") == "text":

print(chunk["content"], end="")

メモリ連携：LangGraph は langgraph-checkpoint-aws パッケージを介して Amazon Bedrock AgentCore の機能である AgentCore Memory と統合されます。これにより、数行のコードで短期間のチェックポイント保存と、インテリジェントな長期メモリの検索が可能になります。

本例で用いる AgentCoreMemorySaver クラスは、ユーザーメッセージや AI の応答、グラフの実行状態、メタデータを含むチェックポイントオブジェクトを管理します。各ノードの処理が完了するたびに LangGraph は自動的にチェックポイントを AgentCore メモリに保存するため、Amazon DynamoDB テーブルの管理や独自シリアライズロジックの実装なしで、状態を保持した対話やワークフローの復元が可能になります。

AgentCoreMemoryStore クラスは、AgentCore が会話から洞察・要約・ユーザーの好意を自動的に抽出するインテリジェントなメモリ機能を提供します。エージェントは将来のやり取りでこれらの記憶を検索できるため、時間とともに改善されるパーソナライズされた体験を実現できます。これは、エージェントがステートレスであるという根本的な課題への解決策です。各対話では新しい状態から始めるのではなく、過去の知識を基に積み上げていくことができます。

import boto3, time

REGION = "us-east-1"

control_client = boto3.client("bedrock-agentcore-control", region_name=REGION)

response = control_client.create_memory(

name="MarketSurveillanceMemory",

description="Memory for market surveillance multi-agent workflow.",

eventExpiryDuration=90, # days

)

MEMORY_ID = response["memory"]["id"]

print(f"Memory ID: {MEMORY_ID}")

ACTIVE を待機する（10 分以内）

通常、作成には 1〜3 分かかります。

deadline = time.time() + 600
while True:
    status = control_client.get_memory(memoryId=MEMORY_ID)["memory"]["status"]
    if status == "ACTIVE":
        break
    if status == "FAILED" or time.time() >= deadline:
        raise RuntimeError(f"Memory {MEMORY_ID} is {status!r}")