Strands Agents のSwarm の協調モードと競争モード

import os
from strands import Agent
from strands.multiagent import Swarm
from strands.models.bedrock import BedrockModel

os.environ["AWS_PROFILE"] = "bedrock"
os.environ["AWS_REGION"] = "us-east-1"

model = BedrockModel(
    model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
)

# ✅ NodeResultオブジェクトから正しくテキストを抽出
def extract(node_result):
    try:
        # NodeResult -> result -> message -> content -> text
        if hasattr(node_result, 'result'):
            agent_result = node_result.result
            if hasattr(agent_result, 'message'):
                message = agent_result.message
                if isinstance(message, dict) and 'content' in message:
                    content = message['content']
                    if isinstance(content, list) and len(content) > 0:
                        if isinstance(content[0], dict) and 'text' in content[0]:
                            text = content[0]['text']
                            if text and text.strip():
                                return text.strip()
    except Exception as e:
        print(f"Extract error: {e}")
    
    return None

# ===== 協調 =====
def collaborative_swarm(task):
    print("=== collaborative_swarm ===")
    
    def mk_agent(name, prompt):
        return Agent(
            name=name,
            system_prompt=(
                prompt
                + "\n重要: handoff禁止。必ず自分で回答せよ。"
            ),
            model=model,
        )
    
    research = mk_agent("research", "生成AIの社会的影響をデータで説明せよ。")
    creative = mk_agent("creative", "生成AIの社会的影響を感情的に説明せよ。")
    critical = mk_agent("critical", "生成AIの社会的課題を論理的に説明せよ。")
    
    swarm = Swarm([research, creative, critical])
    result = swarm(task)
    
    outputs = []
    print("\n--- 個別回答 ---")
    for name, r in result.results.items():
        txt = extract(r)
        if txt:
            outputs.append(f"【{name}の視点】\n{txt}")
            print(f"- {name}: {txt[:100]}...")  # 最初の100文字を表示
    
    # ✅ 空の場合は統合しない
    if not outputs:
        print("(統合できる回答なし)")
        return
    
    # 統合プロンプトを改善
    merge_prompt = f"""以下の3つの異なる視点からの回答を、150文字で論理的に統合してください。

{chr(10).join(outputs)}

重要: handoff禁止。必ず統合された回答を出力せよ。"""
    
    summarizer = mk_agent("summarizer", "与えられた複数の回答を150文字で論理的に統合する専門家")
    merge_swarm = Swarm([summarizer])
    res2 = merge_swarm(merge_prompt)
    
    merged_candidates = [
        extract(x) for x in res2.results.values()
        if extract(x)
    ]
    
    if not merged_candidates:
        print("(統合回答なし)")
        return
    
    print("\n--- 統合結果 ---")
    print(merged_candidates[0])

# ===== 競争 =====
def competitive_swarm(task):
    print("\n=== competitive_swarm ===")
    
    agents = [
        Agent(
            name="research",
            system_prompt="生成AIの社会的影響をデータで説明せよ。handoff禁止。",
            model=model,
        ),
        Agent(
            name="creative",
            system_prompt="生成AIの社会的影響を未来視点で説明せよ。handoff禁止。",
            model=model,
        ),
        Agent(
            name="critical",
            system_prompt="生成AIの社会的リスクを説明せよ。handoff禁止。",
            model=model,
        ),
    ]
    
    swarm = Swarm(agents)
    result = swarm(task)
    
    best_name = None
    best_text = ""
    
    print("\n--- 個別回答 ---")
    for name, r in result.results.items():
        txt = extract(r)
        if not txt:
            continue  # ✅ 空なら表示も判定もしない
        
        print(f"- {name}: {txt[:100]}...")  # 最初の100文字を表示
        
        if len(txt) > len(best_text):
            best_name = name
            best_text = txt
    
    if best_name:
        print("\n=== 勝者 ===")
        print(f"{best_name}（{len(best_text)}文字）")
        print(f"{best_text[:300]}...")  # 最初の300文字を表示
    else:
        print("(勝者なし)")

def main():
    collaborative_swarm("生成AIの社会的影響を説明せよ。")
    competitive_swarm("生成AIの社会的影響を説明せよ。")

if __name__ == "__main__":
    main()

    research = mk_agent("research", "生成AIの社会的影響をデータで説明せよ。")
    creative = mk_agent("creative", "生成AIの社会的影響を感情的に説明せよ。")
    critical = mk_agent("critical", "生成AIの社会的課題を論理的に説明せよ。")
    
    swarm = Swarm([research, creative, critical])

  summarizer = mk_agent("summarizer", "与えられた複数の回答を150文字で論理的に統合する専門家")
    merge_swarm = Swarm([summarizer])
    res2 = merge_swarm(merge_prompt)

 agents = [
        Agent(
            name="research",
            system_prompt="生成AIの社会的影響をデータで説明せよ。handoff禁止。",
            model=model,
        ),
        Agent(
            name="creative",
            system_prompt="生成AIの社会的影響を未来視点で説明せよ。handoff禁止。",
            model=model,
        ),
        Agent(
            name="critical",
            system_prompt="生成AIの社会的リスクを説明せよ。handoff禁止。",
            model=model,
        ),
    ]

        if len(txt) > len(best_text):
            best_name = name
            best_text = txt
    
    if best_name:
        print("\n=== 勝者 ===")
        print(f"{best_name}（{len(best_text)}文字）")
        print(f"{best_text[:300]}...")  # 最初の300文字を表示