第4章リサーチエージェント（Strands Swarm） — 学習メモ¶

書籍「Amazon Bedrock AgentCore実践入門」第4章のサンプルコード（このフォルダ）を読み解いた個人学習メモ。 AgentCore 全体像は ../../lectures/agentcore_basics/STUDY_NOTES.md 参照。実行検証は伴わない。コードに現れた API 名だけ断定し、読めない挙動は「〜と推測」で明示する。

一言で¶

「Orchestrator という受付係の裏に、Plan（戦略立案）→Retrieval/WhatsNew（検索）→Review（採点）の4人チームがいて、採点が70点未満なら Plan にボールを投げ戻す」。この章は書籍中で唯一 bedrock_agentcore を一切 import しない章（chapter4/*.py / chapter4/agents/*.py を grep して確認、ヒット0件）。AgentCore を使わず、素の Strands Agents だけで uv run main.py としてローカルの対話CLIとして完結する。

全体像¶

main.py: while True の対話ループ。Prompt.ask("何でも聞いて下さい") で入力を受け、"exit" を含めば終了、それ以外は orchestrator(user_input) を呼んで結果を rich.Panel + Markdown で表示する（main.py:14-30）。
OrchestratorAgent（agents/orchestrator_agent.py）: モデルは claude-haiku-4-5。ツールは execute_swarm_search の1つだけ（orchestrator_agent.py:96）。additional_request_fields で tool_choice.disable_parallel_tool_use: True を指定（orchestrator_agent.py:90-94）— ツールが1個しかないので並列呼び出し自体が無意味であり、事故防止のための明示設定と読める。
execute_swarm_search ツール本体（orchestrator_agent.py:29-78）は呼ばれるたびに create_plan_agent() / create_retrieval_agent() / create_whatsnew_search_agent() / create_review_agent() で4エージェントを毎回新規生成し、Swarm([plan, retrieval, whatsnew_search, review], entry_point=plan) を組んで stream_async(query) で実行する。
Swarm内の4エージェントは handoff_to_agent(agent_name, message, context) というツールで制御をたらい回しする。このツールはどのエージェントのコードにも定義されていない。ライブラリ側 strands/multiagent/swarm.py の Swarm.__init__（305行目）が _inject_swarm_tools()（520行目）を呼び出し、_create_handoff_tool()（550-581行目）で生成した handoff_to_agent を全ノードの tool_registry に自動登録している（.venv/lib/python3.14/site-packages/strands/multiagent/swarm.py で確認）。だから各 create_*_agent() は tools= にハンドオフ用ツールを一切書いていない。
Swarmは「ノードがハンドオフせずに応答を終えたら完了」という判定をする（swarm.py:819-820 の "no handoff occurred, marking swarm as complete" → Status.COMPLETED）。つまり ReviewAgent がスコア70以上と判断したときは、handoff_to_agent() を呼ばずにただ応答するだけで、Swarm側がそれを検知して打ち切る。逆にスコアが70未満なら handoff_to_agent() で PlanAgent に戻し、再度 Retrieval/WhatsNew → Review のループに入る。
無限ループ防止のデフォルト安全弁がライブラリ側にある: max_handoffs=20 / max_iterations=20 / execution_timeout=900.0秒 / node_timeout=300.0秒（swarm.py:242-246）。本サンプルは Swarm(...) をデフォルト引数のまま生成しているため、これらがそのまま適用される。
全5エージェント共通で ToolLoggingHook（hooks/tool_logging_hook.py）を装着し、BeforeToolCallEvent を購読してツール呼び出しをコンソールにログ出力する。ただし handoff_to_agent だけはログをスキップする（tool_logging_hook.py:22-24）— execute_swarm_search 側の multiagent_handoff イベントで別途 rich.Panel 表示しているため、二重表示を避けていると読める。

flowchart TD
    User(["ユーザー入力"]) --> Orchestrator["OrchestratorAgent<br>claude-haiku-4-5"]
    Orchestrator -->|"execute_swarm_search"| Plan["PlanAgent<br>claude-opus-4-6"]
    Plan -->|"handoff_to_agent"| Retrieval["RetrievalAgent<br>claude-haiku-4-5<br>AWS公式ナレッジMCP"]
    Plan -->|"handoff_to_agent"| WhatsNew["WhatsNewSearchAgent<br>claude-haiku-4-5<br>AWS新着情報RSS"]
    Retrieval -->|"handoff_to_agent"| Review["ReviewAgent<br>claude-sonnet-4-6<br>完全性を0から100点で採点"]
    WhatsNew -->|"handoff_to_agent"| Review
    Review -->|"スコア70点以上・handoffせず終了"| Done["Swarmが完了と判定し<br>Orchestratorへ結果を返却"]
    Review -.->|"スコア70点未満・ギャップを指摘"| Plan

使用ライブラリ・原理¶

pyproject.toml 固定バージョン: strands-agents==1.38.0, strands-agents-tools[rss]==0.5.1, rich==14.3.3（requires-python = ">=3.14"）。dev依存に boto3[crt]==1.42.96（Strands自体はBedrock呼び出しに直接boto3を要求するため、内部依存として引き込まれている）。
strands.multiagent.Swarm: name付き Agent インスタンスをリストで渡すだけで協調が成立する薄いAPI。Graph（有向グラフを明示的に組む）と違い、Swarmは「誰が次を担当するか」をエージェント自身の判断（handoff_to_agent呼び出し）に委ねる自律協調型。
strands.models.BedrockModel + CacheConfig(strategy="auto"): 全5エージェントで指定。プロンプトキャッシュを自動判断で効かせる設定（毎回システムプロンプトが送られるSwarm構成ではキャッシュの効果が大きいと推測）。
strands.tools.mcp.MCPClient + mcp.client.streamable_http.streamable_http_client: RetrievalAgentがAWS公式のナレッジMCPサーバー https://knowledge-mcp.global.api.aws にStreamable HTTPで接続する（retrieval_agent.py:28-31）。
strands.tools.executors.SequentialToolExecutor: RetrievalAgentに指定（retrieval_agent.py:42）。ツール呼び出しを並列でなく順番に実行させる指定で、MCP経由の検索を1件ずつ処理させる意図と読める。
strands_tools.rss / strands_tools.current_time: WhatsNewSearchAgentがAWS What's New RSSフィード検索と現在時刻取得に使用（whatsnew_search_agent.py:4,27）。strands-agents-tools[rss] のextra指定が必要（pyproject.toml:10）。
strands.hooks.HookProvider / HookRegistry / BeforeToolCallEvent: フック機構。register_hooks() でイベント購読を宣言し、Agent生成時に hooks=[ToolLoggingHook()] で注入する。

ファイル別の役割¶

ファイル	役割
`main.py`	対話ループ本体。`Prompt.ask` で入力受付、`orchestrator(user_input)` 呼び出し、`rich.Panel`+`Markdown` で結果表示
`agents/orchestrator_agent.py`	`OrchestratorAgent` 生成。唯一のツール `execute_swarm_search` の中で `Swarm` を組み立てて `stream_async` 実行し、最終ノードの結果を取り出す
`agents/plan_agent.py`	`PlanAgent` 生成。ユーザーのクエリを分析し検索戦略を立案、`handoff_to_agent()` で Retrieval/WhatsNew に振り分ける（システムプロンプトのみでツール定義は無し。ハンドオフはSwarmが自動注入）
`agents/retrieval_agent.py`	`RetrievalAgent` 生成。`MCPClient(streamable_http_client(...))` で AWS公式ナレッジMCP `https://knowledge-mcp.global.api.aws` を叩く。`tool_executor=SequentialToolExecutor()` で逐次実行
`agents/whatsnew_search_agent.py`	`WhatsNewSearchAgent` 生成。`strands_tools.rss` で AWS What's New RSS（`https://aws.amazon.com/about-aws/whats-new/recent/feed/`）を検索。`current_time` も併用
`agents/review_agent.py`	`ReviewAgent` 生成。情報の完全性を0-100で採点し、70未満なら `handoff_to_agent()` で PlanAgent に差し戻す（採点ロジックはシステムプロンプト内の自然言語の指示であり、ライブラリが数値的に強制するものではない）
`hooks/tool_logging_hook.py`	`ToolLoggingHook`。`BeforeToolCallEvent` を購読し、`handoff_to_agent` 以外のツール呼び出しをコンソールへログ出力
`pyproject.toml`	依存バージョン固定（本文「使用ライブラリ・原理」節参照）

中心コードの読み解き¶

execute_swarm_search（orchestrator_agent.py:29-78）を抜粋:

@tool
async def execute_swarm_search(query: str) -> str:
    plan = create_plan_agent()                        # ①
    whatsnew_search = create_whatsnew_search_agent()
    retrieval = create_retrieval_agent()
    review = create_review_agent()

    swarm = Swarm(
        [plan, retrieval, whatsnew_search, review],
        entry_point=plan,
    )                                                  # ②

    multiagent_result: SwarmResult = None

    async for event in swarm.stream_async(query):      # ③
        if event.get("type") == "multiagent_result":
            multiagent_result = event.get("result")    # ④

    final_node_id = multiagent_result.node_history[-1].node_id  # ⑤
    final_result = multiagent_result.results[final_node_id]
    return final_result.result.message

番号	行	やってること	なぜ
①	orchestrator_agent.py:33-36	ツール呼び出しのたびに4エージェントを新規生成	Swarmはノードの実行状態を内部に持つため、リクエストごとにクリーンな状態で開始する。ライブラリ側もセッション永続を明示的に未サポートとしている（`swarm.py:517` の `_validate_swarm` が `node._session_manager is not None` を弾く）
②	orchestrator_agent.py:38-41	`Swarm([...], entry_point=plan)` でチーム結成	name付き `Agent` をリストで渡すだけで協調が成立する。エッジを明示的に定義するGraphと違い、誰が次を担当するかはエージェント自身の`handoff_to_agent`呼び出しに委ねる
③	orchestrator_agent.py:45	`stream_async(query)` でイベント駆動実行	`multiagent_node_start` / `multiagent_handoff` / `multiagent_node_stop` / `multiagent_result` を見ながらrichでリアルタイム表示するため
④	orchestrator_agent.py:69-71	`multiagent_result` に最終結果を保持	`multiagent_result` イベントが発火するのはSwarm全体が完了（`Status.COMPLETED`）した時の1回だけと読める
⑤	orchestrator_agent.py:74-76	`node_history[-1]` から最後に実行したノードの結果を取得	Swarmは「ハンドオフせず応答を終えたノード」で打ち切るため、`node_history`の末尾＝実際に最終回答を確定させたエージェント（通常はReviewAgent。スコア70以上でhandoffしなかった回）

学んだこと（要点）¶

Swarmの協調はエージェント任せの自律型。LangGraphのmulti_agent lecture（Supervisor/Swarm/Command(goto=)）と比べ、Strandsは「name付きAgentのリストを渡すだけ」で成立し、遷移ロジック自体（handoff_to_agentツール）はライブラリが自動注入する。書き味は圧倒的に薄い。
完了判定はシステムプロンプトの自然言語指示とライブラリの機構が組み合わさって成立している。「70点以上なら終了」という数値基準はどこにも構造的なコードとして存在せず、ReviewAgentのLLMが「handoff_to_agentを呼ばない」という行動を選ぶことでのみ実現される。つまりLLMの指示追従性に完全に依存した設計。
モデルの使い分けがコスト最適化になっている: 実行頻度が高い/判断が軽いエージェント（Orchestrator, Retrieval, WhatsNew）はhaiku、難しい戦略立案（Plan）はopus、品質判定（Review）はsonnetと、役割の重さに応じて3段階のモデルを配分している。
MCPは「AWS公式が用意した既存サーバーに繋ぐだけ」。第8章（Gateway）のような自前MCPサーバー構築とは違い、RetrievalAgentはhttps://knowledge-mcp.global.api.awsという既製のエンドポイントにstreamable_http_clientで繋ぐだけで、AWS公式ドキュメント検索能力を獲得している。

落とし穴・現代版に移植するなら¶

AWS公式MCPサーバーのURLがハードコード（retrieval_agent.py:30）。本番なら環境変数化・設定ファイル化すべき箇所（書籍の初心者向けコードなので簡潔さ優先と推測）。
Swarmの安全弁（max_handoffs=20等）はデフォルト値のまま。ReviewAgentとPlanAgentが噛み合わずに何度も往復するケースを想定するなら、execution_timeoutやrepetitive_handoff_detection_windowを明示的に設定してループ検知を強めた方が安全。
callback_handler=Noneが全エージェントに設定されている（例: orchestrator_agent.py:97）ため、Strands標準の逐次出力コールバックは無効化されており、表示はすべてstream_asyncのイベントハンドリングとToolLoggingHook側で自前実装している。この構成を真似るなら、両方の仕組みを理解しないと「何も表示されない」状態にハマる。
disable_parallel_tool_use: True（orchestrator_agent.py:90-94）はOrchestratorAgentにしか付いていない。ツールを複数持つエージェントに拡張する場合は、本当に並列不可でよいかを見直す必要がある。

記事参照¶

書籍「Amazon Bedrock AgentCore実践入門」第4章。
関連: ../../lectures/agentcore_basics/STUDY_NOTES.md（AgentCore全体像・第4章の位置づけ）
関連: ~/ai-engineering-study/lectures/multi_agent/（Supervisor/Swarmのlecture。LangGraph版との対比）

作成: 2026-07-17 ／最終更新: 2026-07-17

第4章 リサーチエージェント（Strands Swarm） — 学習メモ¶