Strands Agents のSwarm （複数エージェントの同時実行）を試す

# swarm.py
import asyncio
import os
from strands import Agent
from strands.multiagent import Swarm
from strands.models.bedrock import BedrockModel

# --- AWS プロファイルとリージョン設定 ---
os.environ["AWS_PROFILE"] = "bedrock"
os.environ["AWS_REGION"] = "us-east-1"

# --- Bedrock Claude モデル ---
bedrock_model = BedrockModel(
    model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
)

# --- エージェント定義 ---
market_agent = Agent(
    system_prompt=(
        "私は market_agent です。必ず出力の最初に 'market_agent:' と書いてください。"
        "生成AIの市場動向を分析し、日本語で200文字以内にまとめてください。"
    ),
    name="market_agent",
    model=bedrock_model,
)

tech_agent = Agent(
    system_prompt=(
        "私は tech_agent です。必ず出力の最初に 'tech_agent:' と書いてください。"
        "生成AIの技術的側面（仕組み・課題・進化）を日本語で200文字以内にまとめてください。"
    ),
    name="tech_agent",
    model=bedrock_model,
)

biz_agent = Agent(
    system_prompt=(
        "私は biz_agent です。必ず出力の最初に 'biz_agent:' と書いてください。"
        "生成AIのビジネス応用事例や効果を日本語で200文字以内にまとめてください。"
    ),
    name="biz_agent",
    model=bedrock_model,
)

summary_agent = Agent(
    system_prompt=(
        "私は summary_agent です。必ず出力の最初に 'summary_agent:' と書いてください。"
        "market, tech, biz 各エージェントの情報を統合して、"
        "日本語で200文字以内に簡潔にまとめてください。"
    ),
    name="summary_agent",
    model=bedrock_model,
)

# --- Swarm 作成 ---
swarm = Swarm(
    [market_agent, tech_agent, biz_agent, summary_agent],
    max_handoffs=20,
    max_iterations=20,
    execution_timeout=900.0,
    node_timeout=300.0,
)

def main():
    # ✅ Swarm 実行（同期呼び出し）
    result = swarm(
        "生成AIの市場・技術・ビジネス活用について分析し、日本語で200文字以内に要約してください。"
    )

    print("\n\n=== 実行結果 ===")
    print(f"ステータス: {result.status}")

    # --- summary_agent の最終要約 ---
    if "summary_agent" in result.results:
        print("\n--- 最終要約 (200文字以内) ---")
        sa = result.results["summary_agent"]
        final_text = (
            getattr(sa, "result", None)
            or getattr(sa, "final_output", None)
            or getattr(sa, "content", None)
            or getattr(sa, "output", None)
        )
        print(final_text if final_text else "（出力なし）")

if __name__ == "__main__":
    main()

swarm = Swarm(
    [market_agent, tech_agent, biz_agent, summary_agent],
    max_handoffs=20,
    max_iterations=20,
    execution_timeout=900.0,
    node_timeout=300.0,
)
<snip>
    result = swarm(
        "生成AIの市場・技術・ビジネス活用について分析し、日本語で200文字以内に要約してください。"
    )

market_agent = Agent(
    system_prompt=(
        "私は market_agent です。必ず出力の最初に 'market_agent:' と書いてください。"
        "生成AIの市場動向を分析し、日本語で200文字以内にまとめてください。"
    ),
    name="market_agent",
    model=bedrock_model,
)