Strands Agents SDK マルチエージェント — Swarm と Graph をネストして組み合わせる

from strands import Agent
from strands.models import BedrockModel
from strands.multiagent import Swarm, GraphBuilder
 
bedrock_model = BedrockModel(
    model_id="us.anthropic.claude-sonnet-4-20250514-v1:0",
    region_name="us-east-1",
)
 
medical_researcher = Agent(
    name="medical_researcher", model=bedrock_model,
    system_prompt="You are a medical research specialist. Research the medical aspects and hand off to another researcher.",
    callback_handler=None,
)
tech_researcher = Agent(
    name="tech_researcher", model=bedrock_model,
    system_prompt="You are a technology research specialist. Research the technology aspects. Provide your findings without handing off.",
    callback_handler=None,
)
research_swarm = Swarm(
    [medical_researcher, tech_researcher],
    entry_point=medical_researcher,
    max_handoffs=5, max_iterations=5,
)
 
report_writer = Agent(
    name="report_writer", model=bedrock_model,
    system_prompt="You are a report writing specialist. Write a concise 2-3 sentence summary combining all research.",
    callback_handler=None,
)
 
builder = GraphBuilder()
builder.add_node(research_swarm, "research_team")
builder.add_node(report_writer, "report")
builder.add_edge("research_team", "report")
builder.set_entry_point("research_team")
 
graph = builder.build()
result = graph("What is the impact of AI on healthcare?")
 
print(f"Status: {result.status}")
print(f"Execution order: {[node.node_id for node in result.execution_order]}")
 
for node_id, node_result in result.results.items():
    print(f"\n--- {node_id} ---")
    print(f"  Status: {node_result.status}")
    print(f"  Time: {node_result.execution_time}ms")
    if node_result.result and hasattr(node_result.result, 'node_history'):
        print(f"  Swarm history: {[n.node_id for n in node_result.result.node_history]}")
    if node_result.result and hasattr(node_result.result, 'message') and node_result.result.message:
        for block in node_result.result.message.get('content', []):
            if 'text' in block:
                print(f"  Output: {block['text'][:150]}...")
                break

from strands import Agent
from strands.models import BedrockModel
from strands.multiagent import GraphBuilder
from strands.hooks import BeforeNodeCallEvent, AfterNodeCallEvent
 
bedrock_model = BedrockModel(
    model_id="us.anthropic.claude-sonnet-4-20250514-v1:0",
    region_name="us-east-1",
)
 
researcher = Agent(
    name="researcher", model=bedrock_model,
    system_prompt="You are a research specialist. Provide key facts in bullet points.",
    callback_handler=None,
)
writer = Agent(
    name="writer", model=bedrock_model,
    system_prompt="You are a writing specialist. Write a 2-sentence summary.",
    callback_handler=None,
)
 
builder = GraphBuilder()
builder.add_node(researcher, "research")
builder.add_node(writer, "report")
builder.add_edge("research", "report")
builder.set_entry_point("research")
graph = builder.build()
 
def on_node_start(event: BeforeNodeCallEvent) -> None:
    print(f"[HOOK] Node starting: {event.node_id}")
 
def on_node_end(event: AfterNodeCallEvent) -> None:
    print(f"[HOOK] Node completed: {event.node_id}")
 
graph.hooks.add_callback(BeforeNodeCallEvent, on_node_start)
graph.hooks.add_callback(AfterNodeCallEvent, on_node_end)
 
result = graph("What is Amazon S3?")
print(f"\nStatus: {result.status}")
print(f"Execution order: {[node.node_id for node in result.execution_order]}")

from strands import Agent, tool, ToolContext
from strands.models import BedrockModel
from strands.multiagent import Swarm
 
bedrock_model = BedrockModel(
    model_id="us.anthropic.claude-sonnet-4-20250514-v1:0",
    region_name="us-east-1",
)
 
@tool(context=True)
def check_permissions(action: str, tool_context: ToolContext) -> str:
    """Check if the current user has permission to perform an action.
 
    Args:
        action: The action to check permissions for
 
    Returns:
        str: Permission check result
    """
    user_id = tool_context.invocation_state.get("user_id", "unknown")
    role = tool_context.invocation_state.get("role", "guest")
    result = f"User {user_id} (role: {role}) - permission for '{action}': {'granted' if role == 'admin' else 'denied'}"
    print(f"[TOOL] {result}")
    return result
 
researcher = Agent(
    name="researcher", model=bedrock_model,
    system_prompt="Check permissions for 'read_data' first, then research the topic briefly. Hand off to the writer when done.",
    tools=[check_permissions], callback_handler=None,
)
writer = Agent(
    name="writer", model=bedrock_model,
    system_prompt="Check permissions for 'write_report' first, then write a 1-sentence summary. Do not hand off.",
    tools=[check_permissions], callback_handler=None,
)
 
swarm = Swarm([researcher, writer], entry_point=researcher, max_handoffs=5, max_iterations=5)
 
print("=== Admin user ===")
result1 = swarm(
    "What is Amazon S3? Keep it brief.",
    invocation_state={"user_id": "user-123", "role": "admin"},
)
print(f"Status: {result1.status}")
 
print("\n=== Guest user ===")
swarm2 = Swarm(
    [
        Agent(name="r", model=bedrock_model, system_prompt="Check permissions for 'read_data' first, then research briefly. Hand off to writer.", tools=[check_permissions], callback_handler=None),
        Agent(name="w", model=bedrock_model, system_prompt="Check permissions for 'write_report' first, then write a 1-sentence summary. Do not hand off.", tools=[check_permissions], callback_handler=None),
    ],
    max_handoffs=5, max_iterations=5,
)
result2 = swarm2(
    "What is Amazon S3? Keep it brief.",
    invocation_state={"user_id": "user-456", "role": "guest"},
)
print(f"Status: {result2.status}")