Strands Agents SDK マルチエージェント — Graph でワークフローを構造化する

from strands import Agent
from strands.models import BedrockModel
from strands.multiagent import GraphBuilder
 
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 about the given topic in bullet points.",
    callback_handler=None,
)
analyst = Agent(
    name="analyst", model=bedrock_model,
    system_prompt="You are an analysis specialist. Analyze the research provided and identify the top 3 insights.",
    callback_handler=None,
)
report_writer = Agent(
    name="report_writer", model=bedrock_model,
    system_prompt="You are a report writing specialist. Write a concise 2-3 sentence summary based on the analysis.",
    callback_handler=None,
)
 
builder = GraphBuilder()
builder.add_node(researcher, "research")
builder.add_node(analyst, "analysis")
builder.add_node(report_writer, "report")
builder.add_edge("research", "analysis")
builder.add_edge("analysis", "report")
builder.set_entry_point("research")
 
graph = builder.build()
result = graph("What is Amazon Bedrock?")
 
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")

from strands import Agent
from strands.models import BedrockModel
from strands.multiagent import GraphBuilder
 
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 about the given topic.",
    callback_handler=None,
)
analyst = Agent(
    name="analyst", model=bedrock_model,
    system_prompt="You are an analysis specialist. Analyze the research and identify key insights.",
    callback_handler=None,
)
fact_checker = Agent(
    name="fact_checker", model=bedrock_model,
    system_prompt="You are a fact-checking specialist. Verify the claims in the research and note any issues.",
    callback_handler=None,
)
report_writer = Agent(
    name="report_writer", model=bedrock_model,
    system_prompt="You are a report writing specialist. Write a concise summary combining the analysis and fact-check results.",
    callback_handler=None,
)
 
builder = GraphBuilder()
builder.add_node(researcher, "research")
builder.add_node(analyst, "analysis")
builder.add_node(fact_checker, "fact_check")
builder.add_node(report_writer, "report")
builder.add_edge("research", "analysis")
builder.add_edge("research", "fact_check")
builder.add_edge("analysis", "report")
builder.add_edge("fact_check", "report")
builder.set_entry_point("research")
 
graph = builder.build()
result = graph("What is Amazon Bedrock?")
 
print(f"Status: {result.status}")
print(f"Execution order: {[node.node_id for node in result.execution_order]}")
print(f"Total nodes: {result.total_nodes}")
print(f"Completed nodes: {result.completed_nodes}")
 
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")

from strands import Agent
from strands.models import BedrockModel
from strands.multiagent import GraphBuilder
from strands.multiagent.graph import GraphState
 
bedrock_model = BedrockModel(
    model_id="us.anthropic.claude-sonnet-4-20250514-v1:0",
    region_name="us-east-1",
)
 
classifier = Agent(
    name="classifier", model=bedrock_model,
    system_prompt="You are a classifier. Classify the input as either 'technical' or 'business'. Reply with ONLY the word 'technical' or 'business'.",
    callback_handler=None,
)
tech_specialist = Agent(
    name="tech_specialist", model=bedrock_model,
    system_prompt="You are a technical specialist. Provide a technical explanation in 2-3 sentences.",
    callback_handler=None,
)
business_specialist = Agent(
    name="business_specialist", model=bedrock_model,
    system_prompt="You are a business specialist. Explain the business impact in 2-3 sentences.",
    callback_handler=None,
)
 
def is_technical(state: GraphState) -> bool:
    result = state.results.get("classifier")
    if not result or not result.result:
        return False
    for block in result.result.message.get('content', []):
        if 'text' in block and 'technical' in block['text'].lower():
            return True
    return False
 
def is_business(state: GraphState) -> bool:
    result = state.results.get("classifier")
    if not result or not result.result:
        return False
    for block in result.result.message.get('content', []):
        if 'text' in block and 'business' in block['text'].lower():
            return True
    return False
 
builder = GraphBuilder()
builder.add_node(classifier, "classifier")
builder.add_node(tech_specialist, "tech")
builder.add_node(business_specialist, "business")
builder.add_edge("classifier", "tech", condition=is_technical)
builder.add_edge("classifier", "business", condition=is_business)
builder.set_entry_point("classifier")
 
graph = builder.build()
 
print("=== Technical question ===")
result1 = graph("How does container orchestration work in Kubernetes?")
print(f"Execution order: {[node.node_id for node in result1.execution_order]}")
 
print("\n=== Business question ===")
graph2 = builder.build()
result2 = graph2("What is the ROI of migrating to cloud computing?")
print(f"Execution order: {[node.node_id for node in result2.execution_order]}")

from strands import Agent
from strands.models import BedrockModel
from strands.multiagent import GraphBuilder
from strands.multiagent.graph import GraphState
 
bedrock_model = BedrockModel(
    model_id="us.anthropic.claude-sonnet-4-20250514-v1:0",
    region_name="us-east-1",
)
 
draft_writer = Agent(
    name="draft_writer", model=bedrock_model,
    system_prompt="You are a draft writer. Write or revise a haiku about programming based on the feedback provided. Output ONLY the haiku (3 lines).",
    callback_handler=None,
)
reviewer = Agent(
    name="reviewer", model=bedrock_model,
    system_prompt="""You are a strict haiku reviewer. Check if the text is a valid haiku (5-7-5 syllable pattern).
If it needs revision, start your response with 'REVISION NEEDED:' followed by specific feedback.
If it's approved, start your response with 'APPROVED:' followed by a brief comment.""",
    callback_handler=None,
)
publisher = Agent(
    name="publisher", model=bedrock_model,
    system_prompt="You are a publisher. Format the approved haiku nicely with a title.",
    callback_handler=None,
)
 
def needs_revision(state: GraphState) -> bool:
    result = state.results.get("reviewer")
    if not result or not result.result:
        return False
    for block in result.result.message.get('content', []):
        if 'text' in block and 'REVISION NEEDED' in block['text'].upper():
            return True
    return False
 
def is_approved(state: GraphState) -> bool:
    result = state.results.get("reviewer")
    if not result or not result.result:
        return False
    for block in result.result.message.get('content', []):
        if 'text' in block and 'APPROVED' in block['text'].upper():
            return True
    return False
 
builder = GraphBuilder()
builder.add_node(draft_writer, "draft_writer")
builder.add_node(reviewer, "reviewer")
builder.add_node(publisher, "publisher")
builder.add_edge("draft_writer", "reviewer")
builder.add_edge("reviewer", "draft_writer", condition=needs_revision)
builder.add_edge("reviewer", "publisher", condition=is_approved)
builder.set_entry_point("draft_writer")
builder.set_max_node_executions(10)
builder.set_execution_timeout(300)
builder.reset_on_revisit(True)
 
graph = builder.build()
result = graph("Write a haiku about Python programming")
 
print(f"Status: {result.status}")
print(f"Execution order: {[node.node_id for node in result.execution_order]}")
print(f"Total executions: {len(result.execution_order)}")
 
for node_id, node_result in result.results.items():
    print(f"\n--- {node_id} ---")
    print(f"  Status: {node_result.status}")
    if node_result.result and node_result.result.message:
        for block in node_result.result.message.get('content', []):
            if 'text' in block:
                print(f"  Output: {block['text'][:200]}")
                break