【小ネタ】 Strands Agents の Swarm で最終エージェントの回答だけを取り出す

"""Swarm の定義（front → researcher → synthesizer の 3 エージェント構成）"""

from strands import Agent
from strands.multiagent import Swarm

MODEL_ID = "us.anthropic.claude-haiku-4-5-20251001-v1:0"

def create_swarm() -> Swarm:
    front_agent = Agent(
        name="front",
        model=MODEL_ID,
        callback_handler=None,  # デフォルトの標準出力への自動出力を止める
        system_prompt=(
            "あなたは受付エージェントです。"
            "ユーザーからの問い合わせを「問題の要約」「関連情報」「調査してほしいこと」の形に構造化し、"
            "その内容を出力したうえで researcher にハンドオフしてください。"
            "自分では調査や回答を行わないでください。"
        ),
    )

    researcher = Agent(
        name="researcher",
        model=MODEL_ID,
        callback_handler=None,
        system_prompt=(
            "あなたは調査エージェントです。"
            "受け取った内容について、知っている知識をもとに原因候補や関連情報を洗い出してください。"
            "調査の過程や考えたことも出力してかまいません。"
            "調査が終わったら synthesizer にハンドオフしてください。"
        ),
    )

    synthesizer = Agent(
        name="synthesizer",
        model=MODEL_ID,
        callback_handler=None,
        system_prompt=(
            "あなたはレポート作成エージェントです。"
            "これまでの調査内容を、ユーザーに見せる最終レポートとしてまとめてください。"
            "「調査結果まとめ」という見出しで始め、簡潔で読みやすい日本語のレポートだけを出力してください。"
            "他のエージェントへのハンドオフは不要です。"
        ),
    )

    return Swarm(
        [front_agent, researcher, synthesizer],
        entry_point=front_agent,
        max_handoffs=10,
        max_iterations=15,
    )

USER_INPUT = (
    "Web アプリのレスポンスが昨日から急に遅くなりました。"
    "特に一覧画面の表示に 10 秒以上かかります。データベースは RDS (PostgreSQL) です。"
    "考えられる原因を調べてまとめてください。"
)

"""全エージェントの発話をそのまま連結する例

協調プロセス（受付の構造化・ハンドオフの口上・調査の実況）を
すべて見せたい場合はこの形で OK。
最終レポートだけを見せたい場合は final_only.py の形にする。
"""

import asyncio

from swarm_def import USER_INPUT, create_swarm

async def main() -> None:
    swarm = create_swarm()
    final_text = ""

    async for event in swarm.stream_async(USER_INPUT):
        if event.get("type") == "multiagent_node_stream":
            inner = event.get("event", {})
            if isinstance(inner, dict) and "data" in inner:
                text = inner["data"] if isinstance(inner["data"], str) else ""
                if text:
                    final_text += text

    print("=" * 60)
    print("回答（全エージェントの発話をそのまま連結）")
    print("=" * 60)
    print(final_text)

if __name__ == "__main__":
    asyncio.run(main())

"""改善版: テキストをノード別に蓄積し、最終エージェント（synthesizer）の分だけを採用する

multiagent_node_stream イベントはトップレベルに node_id を持つため、
それをキーにノード別の dict へ蓄積する。
synthesizer に到達せず終了した場合は、最後に発話したノードの出力で代替する。
"""

import asyncio
import logging

from swarm_def import USER_INPUT, create_swarm

logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger(__name__)

FINAL_NODE = "synthesizer"

async def main() -> None:
    swarm = create_swarm()

    node_texts: dict[str, str] = {}
    last_text_node = ""

    async for event in swarm.stream_async(USER_INPUT):
        if event.get("type") == "multiagent_node_stream":
            node_id = event.get("node_id", "")
            inner = event.get("event", {})
            if isinstance(inner, dict) and "data" in inner:
                text = inner["data"] if isinstance(inner["data"], str) else ""
                if text:
                    node_texts[node_id] = node_texts.get(node_id, "") + text
                    last_text_node = node_id

    # 最終レポート担当（synthesizer）の出力だけを採用
    final_text = node_texts.get(FINAL_NODE, "").strip()

    # synthesizer まで到達せずに終了した場合のフォールバック
    if not final_text and last_text_node:
        final_text = node_texts.get(last_text_node, "").strip()

    # デバッグ用: どのエージェントが何文字話したか
    logger.info(f"nodes: { {k: len(v) for k, v in node_texts.items()} }")

    print("=" * 60)
    print("最終回答（synthesizer のレポートのみ）")
    print("=" * 60)
    print(final_text)

if __name__ == "__main__":
    asyncio.run(main())

uv init
uv add "strands-agents==1.47.0"
AWS_REGION=us-west-2 uv run stream_all.py
AWS_REGION=us-west-2 uv run final_only.py