This content originally appeared on Level Up Coding – Medium and was authored by Itsuki
Scrambled Swarm! Structured Graph! One step closer to making a Self-automated Agent?
If you get a chance to read my previous article Strands Agent: All The Basics + Key Features I Use/Like the MOST, just like the title suggests, Strands Agents is, By Far, My Favorite Framework for building an Agent!
And here we go! Another really awesome point about this framework!
Strands provides couple pretty unique and interesting (of course, in MY opinion) ways in implementing multi-agent in addition to the traditional ones!
A scrambled Swarm! A structured Graph!
(However, it doesn’t have one of the common ones that other agent frameworks have, that complete hands-off!)
Anyway!
Let’s check those out!
How we can build it, when it might be useful, and some of my ways on using those!
Agent As Tools
This one is obviously not the interesting one I am talking about!
It is the traditional hierarchical multi-agent systems, but I decided to put it here just so that we can see how the other two are different!
I am pretty sure we all know by now what this Agents as Tools pattern is all about!
We have a bunch of sub-agents that is modeled as tools to perform domain-specific tasks, and an Supervisor (or Orchestrator) agent that is responsible for
- handling user interaction
- determining which specialized agent to call and calling those (or not)
- coordinating a final response using the response from the sub agents
To implement it, all we have to do is to define individual assistants as a tool with the tool decorator, just like the name suggests!
from strands import tool
from strands import Agent
from strands.models import BedrockModel
region = "us-east-1"
bedrock_model = BedrockModel(
model_id="us.anthropic.claude-sonnet-4-20250514-v1:0",
region_name=region,
)
@tool
def pokemon_battle_assistant(query: str) -> str:
"""
Give pokemon battle related advice.
Args:
query: Some questions or requests regarding to pokemon battle.
Returns:
A detailed battling advice
"""
try:
pokemon_agent = Agent(
system_prompt="You specialized in giving pokemon battle related advices. ",
model=bedrock_model
)
response = pokemon_agent(query)
return str(response)
except Exception as e:
return f"Error in research assistant: {str(e)}"
@tool
def fgo_assistant(query: str) -> str:
"""
Give FGO (mobile game) related advice.
Args:
query: Some questions or requests regarding to FGO (mobile game) .
Returns:
A detailed battling advice
"""
try:
fgo_agent = Agent(
system_prompt="You specialized in giving FGO (mobile game) related advices. ",
model=bedrock_model
)
response = fgo_agent(query)
return str(response)
except Exception as e:
return f"Error in research assistant: {str(e)}"
orchestrator_system_prompt = """
You are an assistant that routes user's request to specialized agents:
- For pokmeon battle → Use the pokemon_battle_assistant tool
- For FGO (mobile game) gameplay → Use the fgo_assistant tool
Always select the most appropriate tool based on the user's query.
"""
orchestrator_agent = Agent(
system_prompt=orchestrator_system_prompt,
callback_handler=None,
tools=[pokemon_battle_assistant, fgo_assistant],
model=bedrock_model
)
Swarm
This is the first interesting Multi-Agent Pattern I would like to share today!
A Swarm is a collaborative agent orchestration system where multiple agents work together as a team to solve complex task, and most importantly those agents share context and working memory!
I call it a scramble because each agent, also know as nodes, in a Swarm
- Has access to the full task context
- Can see the history of which agents have worked on the task
- Can access shared knowledge contributed by other agents
- Can decide when to hand off to another agent with different expertise
That is!
Autonomous agent collaboration without central control!
Let’s implement one ourselves really quick!
Static Swarm
What do I meant by static here?
We create the agents and the Swarm workflow ahead of time!
Let’s start with creating a number of agents that is specialized in different areas.
# Create individual specialized agents
from strands_tools import http_request
pokemon_helper = Agent(
model=bedrock_model,
name="pokemon_helper",
system_prompt="You are a pokmeon specialist answering questions on pokemons",
)
pokemon_battle_specialist = Agent(
model=bedrock_model,
name="pokemon_battle_specialist",
system_prompt="You are a pokmeon battle specialist answer question on pokemon battles."
)
pokemon_info_specialist = Agent(
model=bedrock_model,
name="pokemon_info_specialist",
system_prompt="You are knowledged in pokemon details such as it's type, skills, and etc."
)
pokemon_web_reasearcher = Agent(
model=bedrock_model,
name="pokemon_web_reasearcher",
system_prompt="You are specialized in looking for pokmeon related information over the web.",
tools=[http_request]
)
We will then create a Swarm with the agents above.
from strands.multiagent import Swarm
swarm = Swarm(
# first agent will be used as the entry point
[pokemon_helper, pokemon_battle_specialist, pokemon_info_specialist, pokemon_web_reasearcher],
max_handoffs=20,
max_iterations=20,
execution_timeout=900.0, # in seconds, ie: 15 minutes
node_timeout=300.0, # in seconds, ie: 5 minutes per agent
# Set appropriate values for repetitive_handoff_detection_window and repetitive_handoff_min_unique_agents to prevent ping-pong behavior
repetitive_handoff_detection_window=8, # There must be >= 3 unique agents in the last 8 handoffs
repetitive_handoff_min_unique_agents=3
)
The first agent we passed into the nodes list will be the entry point.
In addition, we have also specified couple other parameters.
- max_handoffs: Maximum number of agent handoffs allowed. Default to 20.
- max_iterations: Maximum total iterations across all agents. Default to 20.
- execution_timeout: Total execution timeout in seconds. Default to 900.0 seconds (15 min).
- node_timeout: Individual agent timeout in seconds. Default to 300.0 seconds (5 min).
- repetitive_handoff_detection_window: Number of recent nodes to check for ping-pong behavior. Default to 0, ie: check is disabled.
- repetitive_handoff_min_unique_agents: Minimum unique nodes required in recent sequence. Default to 0, ie: not required.
Note that it is indeed recommended to set some values for repetitive_handoff_detection_window and repetitive_handoff_min_unique_agents to prevent ping-pong behavior!
What happens when create a Swarm is that each agent is automatically equipped with a special handoff_to_agent tool that transfers control to another agent in the swarm for specialized help.
Which means!
Trying to create Swarm on the same set of agents without re-creating the agents will cause error!
Because! Same tool, ie: that handoff_to_agent, cannot be registered twice on the same agent!
We can then invoke the Swarm just like we do for regular agents.
response = swarm("help me choose a pokemon to fight against water type pokemons and the skills I should use. I would also like to know how other people uses this pokemon.")
# Reponse:
# I'd be happy to help you choose a Pokémon to fight against Water-type Pokémon! This is a great strategic question that involves understanding type effectiveness and move selection.
# Let me hand this off to our pokemon_battle_specialist who has deep expertise in battle strategy and type matchups.
# Tool #1: handoff_to_agent
# Perfect! I've handed you off to our pokemon_battle_specialist who will provide you with expert advice on the best Pokémon choices and battle strategies to effectively counter Water-type Pokémon. They'll cover type advantages, specific Pokémon recommendations, and the most effective moves to use in these matchups.Great! I'll help you choose the best Pokémon and moves to counter Water-type opponents. As a Pokémon battle specialist, here's my comprehensive advice:
# ...
print(f"Status: {response.status}")
# Status: Status.COMPLETED
print(f"Node history: {[node.node_id for node in response.node_history]}")
# Node history: ['pokemon_helper', 'pokemon_battle_specialist']And of course, multi-modal inputs like text and images using ContentBlocks is also supported!
Dynamic Swarm
A dynamic swarm, ie: agents generated on the runtime based on user’s query, can be achieved by using the swarm tool available in the Strands tools.
from strands_tools import swarm
swarm_creater_agent = Agent(
model=bedrock_model,
tools=[swarm],
system_prompt="Create a swarm of agents to solve the user's query."
)
swarm_creater_agent("Research and analyze the latest pokemon battle trend in TCG.")
Now!
I personally don’t recommend to make the agent use the swarm tool this way, because since we don’t get to define those configuration parameters like above, it is really easy to go out of control and on forever due to the ping-pong behavior between the agents!
Then when does this tool might be useful?
If you are used to strands tools, and hopefully you are, you know that we can invoke those directly by using agent.tool.tool_name.
And of course, we can do it with this swarm tool as well!
This will allow us to configure the same parameters as above!
result = agent.tool.swarm(
task="Some tasks, ie: user's query",
agents=[
{
"name": "agent_1",
"system_prompt": (
"some_prompt"
),
"tools": ["retrieve", "calculator"],
"model_provider": "bedrock",
"model_settings": {"model_id": "us.anthropic.claude-sonnet-4-20250514-v1:0"}
},
{
"name": "agent_2",
"system_prompt": (
"some_prompt"
),
"tools": ["file_write", "calculator"],
"model_provider": "anthropic",
"model_settings": {"model_id": "claude-sonnet-4-20250514"}
}
],
max_handoffs = 20,
max_iterations = 20
# ... other configuration parameters available as well
)
But this won’t be dynamic any more!
So here is what I do!
- Create a custom swarm tool that wraps around the one provided
- Set the parameter to the function to be only task and agents
- expose the custom one as the tool to the agent
- Call the provided tool (as regular functions!) with the task and agents generated and the configurations of our own choice!
Graph
If what we had above is like a scramble, what we have here, Graph, is a structured, deterministic workflow!
Within a Graph, we have
- Nodes represent agents, custom nodes, or multi-agent systems (Swarm or nested Graphs)
- Edges define the flow (or depnedency ) between nodes. Output from one node becomes input for the dependent nodes.
Now, the Graph in the little image above is acyclic, but Cyclic graph is also supported, which enables feedback loops and iterative refinement workflows! Another little step closer to a self-automated Agent!
Time to make some Graphs ourselves!
Static Graph
Yes, we again have a static version and a dynamic version!
For static graph, let’s build the following!
To build a graph, here are the steps!
- Create individual GraphNode (Agents, Swarms, or nested Graphs)
- Create a GraphBuilder, an interface for constructing graphs.
- Add the GraphNodes to the GraphBuilder
- Define GraphEdge representing the connection between GraphNodes
- Optionally configure the GraphBuilder with some custom parameters such as the entry point for defining the starting nodes for execution, the execution timeout to control the maximum execution time, and etc.
- Build the GraphBuilder to create a Graph
# Create agents (nodes)
pokemon_helper = Agent(
model=bedrock_model,
name="pokemon_helper",
system_prompt="You are a pokmeon specialist answering questions on pokemons",
)
pokemon_card_game_specialist = Agent(
model=bedrock_model,
name="pokemon_card_game_specialist",
system_prompt="You are a pokmeon card game specialist."
)
pokemon_video_game_specialist = Agent(
model=bedrock_model,
name="pokemon_video_game_specialist",
system_prompt="You are a pokmeon video game specialist."
)
pokemon_reporter = Agent(
model=bedrock_model,
name="pokemon_reporter",
system_prompt="You are a report writing specialist summarizing the information you receive and output a report."
)
# Build the graph
from strands.multiagent import GraphBuilder
builder = GraphBuilder()
# Add nodes
builder.add_node(pokemon_helper, "pokemon_helper")
builder.add_node(pokemon_card_game_specialist, "pokemon_card_game_specialist")
builder.add_node(pokemon_video_game_specialist, "pokemon_video_game_specialist")
builder.add_node(pokemon_reporter, "pokemon_reporter")
# Add edges (dependencies)
builder.add_edge("pokemon_helper", "pokemon_card_game_specialist")
builder.add_edge("pokemon_helper", "pokemon_video_game_specialist")
builder.add_edge("pokemon_video_game_specialist", "pokemon_reporter")
builder.add_edge("pokemon_card_game_specialist", "pokemon_reporter")
# Set entry points (optional - will be auto-detected if not specified)
builder.set_entry_point("pokemon_helper")
# Optional: Configure execution limits for safety
builder.set_execution_timeout(600) # 10 minute timeout
# Build the graph
graph = builder.build()
The usage is, again, exactly the same as regular agents, either plain text input, or multi-modal inputs like text and images using ContentBlocks!
response = graph("Research the current pokemon game trend and report")
# # Current Pokémon Game Trends Report (2024)
# ## Recent Major Releases & Performance
# **Pokémon Scarlet and Violet (2022)**
# - Despite technical issues at launch, became one of the fastest-selling Pokémon games
# - Introduced open-world gameplay to the mainline series
# - Over 20 million copies sold worldwide
# ...
def truncate_string(text:str) -> str:
return (text[:100] + '..') if len(text) > 75 else text
# Individaul node results
for node in response.execution_order:
print("---------")
print(node.node_id)
print(node.result)
print(truncate_string(str(node.result.result)))
# ---------
# pokemon_helper
# NodeResult(result=AgentResult(stop_reason='end_turn', message={'role': 'assistant', 'content': [{'text': "# Current Pokémon Game Trends Report (2024)\n\n## Recent Major Releases & Performance\n\n**Pokémon Scarlet and Violet (2022)**\n- Despite technical issues at launch, became one of the fastest-selling Pokémon games\n- Introduced open-world gameplay to the mainline series\n- Over 20 million copies sold worldwide\n\n**The Pokémon Company's Expansion Strategy**\n- DLC content model proven successful with Scarlet/Violet expansions\n- Focus on post-launch content and updates\n\n## Current Gaming Trends in Pokémon\n\n### 1. **Open-World Evolution**\n- Movement away from traditional linear routes\n- Greater exploration freedom\n- Environmental storytelling and discovery-based gameplay\n\n### 2. **Mobile Gaming Dominance**\n- **Pokémon GO** continues strong performance with regular events\n- **Pokémon UNITE** maintains active MOBA player base\n- **Pokémon Sleep** launched in 2023, gamifying sleep tracking\n- **Pokémon TCG Pocket** (2024) bringing card collecting to mobile\n\n### 3. **Multiplayer Focus**\n- Co-op gameplay integration in mainline games\n- Competitive online features\n- Community-driven events and raids\n\n### 4. **Cross-Media Integration**\n- Games tie into anime, TCG, and merchandise\n- Simultaneous releases across different media\n- Unified storylines and character development\n\n## Technical Innovations\n\n- **Quality-of-Life Improvements**: Auto-save, streamlined menus, accessibility options\n- **Visual Upgrades**: Transition to full 3D open environments\n- **Online Features**: Enhanced trading, battling, and social features\n\n## Market Performance Indicators\n\n- Strong digital sales growth\n- Consistent DLC adoption rates\n- Mobile game revenue streams expanding\n- International market expansion in emerging regions\n\n## Looking Ahead\n\nThe franchise appears to be balancing innovation with nostalgia, focusing on accessibility while maintaining core gameplay mechanics that appeal to both new and veteran players.\n\nWould you like me to dive deeper into any specific aspect of these trends?"}]}, metrics=EventLoopMetrics(cycle_count=1, tool_metrics={}, cycle_durations=[12.762686967849731], traces=[<strands.telemetry.metrics.Trace object at 0x114ba7770>], accumulated_usage={'inputTokens': 29, 'outputTokens': 476, 'totalTokens': 505}, accumulated_metrics={'latencyMs': 12147}), state={}), execution_time=12763, status=<Status.COMPLETED: 'completed'>, accumulated_usage={'inputTokens': 29, 'outputTokens': 476, 'totalTokens': 505}, accumulated_metrics={'latencyMs': 12147}, execution_count=1)
# # Current Pokémon Game Trends Report (2024)
# ## Recent Major Releases & Performance
# **Pokémon Scarl..
# ---------
# pokemon_card_game_specialist
# NodeResult(result=AgentResult(stop_reason='end_turn', message={'role': 'assistant', 'content': [{'text': "# Pokémon Trading Card Game Trends Report (2024)\n\nAs a Pokémon TCG specialist, I'll complement the video game trends with current developments in the trading card game space, which often mirrors and influences the broader Pokémon gaming ecosystem.\n\n## Current TCG Market Trends\n\n### **Digital-Physical Integration**\n- **Pokémon TCG Pocket** (2024) is revolutionizing card collecting by bridging physical and digital play\n- QR code integration allowing physical cards to unlock digital content\n- Cross-platform collection tracking and deck building\n\n### **Competitive Scene Evolution**\n- Regional Championships returning to pre-pandemic levels\n- Prize pools increasing significantly (World Championships 2024 exceeded $1M total prizes)\n- Streaming and content creation driving new player acquisition\n\n### **Card Design & Mechanics Trends**\n- **Special Art Rare (SAR)** cards driving collector interest\n- **ex** Pokémon cards returning as the premium card type\n- Rule box mechanics creating more strategic depth\n- Energy acceleration and draw power being carefully balanced\n\n## Market Performance Indicators\n\n### **Collector Market**\n- Alt art and special illustration cards maintaining premium values\n- Japanese sets often driving global trend adoption\n- Vintage card market stabilizing after 2020-2022 boom\n\n### **Accessibility Initiatives**\n- Battle Academy products for new players\n- Simplified starter decks with clear upgrade paths\n- Enhanced online tutorials and learning resources\n\n## Crossover with Video Game Trends\n\nThe TCG directly reflects the video game trends you mentioned:\n- **Scarlet & Violet** TCG series introduced new mechanics paralleling the games\n- Mobile integration supports the broader mobile gaming focus\n- Community events align with video game release schedules\n\n## Strategic Implications\n\nThe Pokémon TCG is positioning itself as both a competitive game and collectible hobby, with digital integration serving as a bridge to attract video game players while maintaining the tactile appeal that distinguishes it from purely digital experiences.\n\nThe synergy between TCG releases and video game launches continues to be a key driver for cross-media engagement and player retention across the entire Pokémon ecosystem."}]}, metrics=EventLoopMetrics(cycle_count=1, tool_metrics={}, cycle_durations=[15.421281099319458], traces=[<strands.telemetry.metrics.Trace object at 0x114d9f110>], accumulated_usage={'inputTokens': 518, 'outputTokens': 476, 'totalTokens': 994}, accumulated_metrics={'latencyMs': 15230}), state={}), execution_time=15421, status=<Status.COMPLETED: 'completed'>, accumulated_usage={'inputTokens': 518, 'outputTokens': 476, 'totalTokens': 994}, accumulated_metrics={'latencyMs': 15230}, execution_count=1)
# # Pokémon Trading Card Game Trends Report (2024)
# As a Pokémon TCG specialist, I'll complement the v..
# ---------
# pokemon_video_game_specialist
# NodeResult(result=AgentResult(stop_reason='end_turn', message={'role': 'assistant', 'content': [{'text': "Based on my analysis of the current Pokémon gaming landscape, here's my assessment of the key trends shaping the franchise:\n\n## Major Gameplay Evolution Trends\n\n**Open-World Transition Success**\nThe shift to open-world gameplay in Scarlet/Violet, despite launch issues, has proven highly successful commercially. This represents the biggest structural change to mainline Pokémon games in decades and signals the franchise's commitment to modernizing core gameplay mechanics.\n\n**Mobile-First Strategy Expansion**\nThe mobile ecosystem is becoming increasingly central to Pokémon's strategy:\n- Pokémon GO's sustained success proves the mobile market's viability\n- New entries like Pokémon Sleep and TCG Pocket show diversification into lifestyle gaming\n- Mobile titles now serve as both revenue generators and player acquisition tools\n\n## Strategic Business Trends\n\n**DLC-Focused Content Model**\nThe franchise has successfully adopted the games-as-a-service approach, extending game lifecycles through substantial DLC content rather than requiring separate sequel purchases.\n\n**Cross-Platform Ecosystem**\nPokémon is building an interconnected ecosystem where games, anime, TCG, and merchandise reinforce each other, creating multiple touchpoints for fan engagement.\n\n## Technical Innovation Patterns\n\n**Quality-of-Life Priority**\nRecent releases show increased focus on accessibility and user experience improvements, suggesting the franchise is prioritizing player retention alongside acquisition.\n\n**Multiplayer Integration**\nEnhanced social features across all platforms indicate recognition that community engagement drives long-term success in modern gaming.\n\n## Market Implications\n\nThe data suggests Pokémon is successfully navigating the challenge of franchise evolution while maintaining its core appeal. The combination of traditional console gaming innovation and aggressive mobile expansion positions them well for continued growth across diverse gaming segments.\n\nThe franchise appears to be in a strong experimental phase, using different platforms to test new gameplay concepts before potentially integrating successful elements into future mainline releases."}]}, metrics=EventLoopMetrics(cycle_count=1, tool_metrics={}, cycle_durations=[18.233130931854248], traces=[<strands.telemetry.metrics.Trace object at 0x11431ad70>], accumulated_usage={'inputTokens': 518, 'outputTokens': 418, 'totalTokens': 936}, accumulated_metrics={'latencyMs': 17648}), state={}), execution_time=18233, status=<Status.COMPLETED: 'completed'>, accumulated_usage={'inputTokens': 518, 'outputTokens': 418, 'totalTokens': 936}, accumulated_metrics={'latencyMs': 17648}, execution_count=1)
# Based on my analysis of the current Pokémon gaming landscape, here's my assessment of the key trends..
# ---------
# pokemon_reporter
# NodeResult(result=AgentResult(stop_reason='end_turn', message={'role': 'assistant', 'content': [{'text': "# Current Pokémon Game Trend Report\n\n## Executive Summary\n\nThe Pokémon franchise is experiencing a significant transformation period in 2024, marked by successful adaptation to modern gaming trends while maintaining its core appeal. The analysis reveals a multi-platform strategy focused on ecosystem integration, quality-of-life improvements, and experimental gameplay approaches across both digital and physical gaming spaces.\n\n## Key Trend Categories\n\n### 1. Platform Evolution and Diversification\n\n**Open-World Gaming Success**\n- Pokémon Scarlet/Violet's open-world format represents the most significant structural change to mainline games in decades\n- Despite technical launch issues, commercial success validates the franchise's modernization efforts\n- Sets foundation for future mainline game development\n\n**Mobile Gaming Expansion**\n- Mobile-first strategy showing strong results with Pokémon GO's sustained success\n- New lifestyle gaming entries (Pokémon Sleep, TCG Pocket) diversifying the mobile portfolio\n- Mobile platforms serving dual purpose as revenue generators and player acquisition tools\n\n### 2. Business Model Innovation\n\n**Games-as-a-Service Adoption**\n- Successful transition to DLC-focused content delivery extending game lifecycles\n- Eliminates need for separate sequel purchases while maintaining revenue streams\n- Allows for deeper content development and player engagement\n\n**Cross-Media Integration**\n- Strengthened connections between video games, trading cards, anime, and merchandise\n- Digital-physical integration in TCG Pocket bridging collection experiences\n- Synchronized release schedules across platforms maximizing cross-promotion opportunities\n\n### 3. Community and Competitive Gaming\n\n**Enhanced Social Features**\n- Multiplayer integration prioritized across all platforms\n- Competitive scenes strengthening with increased prize pools (World Championships 2024 exceeded $1M)\n- Content creation and streaming driving new player acquisition\n\n**Accessibility Focus**\n- Quality-of-life improvements prioritized in recent releases\n- Educational resources and simplified entry products for new players\n- Clear upgrade paths from beginner to advanced gameplay\n\n### 4. Technical and Design Innovation\n\n**Strategic Gameplay Depth**\n- New mechanics in both video games and TCG creating more strategic complexity\n- Energy acceleration and rule box mechanics in cards\n- Enhanced battle systems and multiplayer features in video games\n\n**Collection and Customization**\n- Special Art Rare (SAR) cards and premium collectibles driving engagement\n- Cross-platform collection tracking and deck building\n- Player customization options expanded across platforms\n\n## Market Implications\n\n### Strengths\n- Successfully balancing innovation with franchise tradition\n- Multiple revenue streams reducing platform dependency\n- Strong experimental phase allowing risk-managed innovation testing\n\n### Strategic Positioning\n- Well-positioned for continued growth across diverse gaming segments\n- Ecosystem approach creating multiple player touchpoints\n- Mobile expansion capturing new demographics while retaining core audience\n\n## Future Outlook\n\nThe Pokémon franchise appears to be in a robust experimental phase, using different platforms to test innovative gameplay concepts before integration into future releases. The combination of traditional console gaming evolution and aggressive mobile expansion, supported by physical-digital TCG integration, positions the franchise for sustained growth and market leadership in the evolving gaming landscape.\n\nThe data indicates Pokémon is successfully navigating the challenge of franchise modernization while preserving the core elements that have driven its decades-long success."}]}, metrics=EventLoopMetrics(cycle_count=1, tool_metrics={}, cycle_durations=[18.99148201942444], traces=[<strands.telemetry.metrics.Trace object at 0x1143d0e20>], accumulated_usage={'inputTokens': 958, 'outputTokens': 719, 'totalTokens': 1677}, accumulated_metrics={'latencyMs': 18812}), state={}), execution_time=18992, status=<Status.COMPLETED: 'completed'>, accumulated_usage={'inputTokens': 958, 'outputTokens': 719, 'totalTokens': 1677}, accumulated_metrics={'latencyMs': 18812}, execution_count=1)
# # Current Pokémon Game Trend Report
# ## Executive Summary
# The Pokémon franchise is experiencing a s..
# Overall result
print(f"Status: {response.status}")
# Status: Status.COMPLETED
print(f"Execution order: {[node.node_id for node in response.execution_order]}")
# Execution order: ['pokemon_helper', 'pokemon_card_game_specialist', 'pokemon_video_game_specialist', 'pokemon_reporter']
# Total performance metrics
print(f"Total nodes: {response.total_nodes}")
print(f"Completed nodes: {response.completed_nodes}")
print(f"Failed nodes: {response.failed_nodes}")
print(f"Execution time: {response.execution_time}ms")
print(f"Token usage: {response.accumulated_usage}")
# Total nodes: 4
# Completed nodes: 4
# Failed nodes: 0
# Execution time: 49989ms
# Token usage: {'inputTokens': 2023, 'outputTokens': 2089, 'totalTokens': 4112}Dynamic Graph
Simple!
Use the graph tool in Strands tools to allow Agent to create and orchestrate graphs based on user’s input!
from strands import Agent
from strands_tools import graph
agent = Agent(
model=bedrock_model,
tools=[graph],
system_prompt="Create a graph of agents to solve the user's query."
)
agent("Research the current pokemon game trend and report")
A Little More
Conditional Flow
First of all, if you ever build a flow chart somewhere, it is pretty common to have some if-else!
We can add conditional logic to those edges to achieve this!
For example, we only want to resume the execution, ie: traverse the next node, if nothing failed til this point.
def execute_if_nothing_failed(state: GraphState):
"""Only traverse if there is no failed nodes."""
failed_nodes = state.failed_nodes
return len(failed_nodes) == 0
# usage
builder.add_edge("pokemon_helper", "pokemon_card_game_specialist", condition=execute_if_nothing_failed)
The condition parameter expects a function taking in GraphState as a parameter and returning a bool.
Within the GraphState, we have access to the following parameters.
task: str | list[ContentBlock] = ""
# Execution state
status: Status = Status.PENDING
completed_nodes: set["GraphNode"] = field(default_factory=set)
failed_nodes: set["GraphNode"] = field(default_factory=set)
execution_order: list["GraphNode"] = field(default_factory=list)
start_time: float = field(default_factory=time.time)
# Results
results: dict[str, NodeResult] = field(default_factory=dict)
# Accumulated metrics
accumulated_usage: Usage = field(default_factory=lambda: Usage(inputTokens=0, outputTokens=0, totalTokens=0))
accumulated_metrics: Metrics = field(default_factory=lambda: Metrics(latencyMs=0))
execution_count: int = 0
execution_time: int = 0
# Graph structure info
total_nodes: int = 0
edges: list[Tuple["GraphNode", "GraphNode"]] = field(default_factory=list)
entry_points: list["GraphNode"] = field(default_factory=list)
Custom Node Type
There are a bunch of built-in types that we can call add_node on, Agents, Other Graphs, Swarms!
However, if the provided ones doesn’t meet your needs, it is also possible to create custom node types by extending MultiAgentBase to implement deterministic business logic, data processing pipelines, and hybrid workflows.
Let’s create one ourselves just to get an idea of how custom nodes can be made!
from strands.multiagent.base import MultiAgentBase, NodeResult, Status, MultiAgentResult
from strands.agent.agent_result import AgentResult
import random
class RandomlyFailedAgentNode(MultiAgentBase):
"""Execute deterministic Python functions as graph nodes."""
def __init__(self, agent: Agent, name: str):
super().__init__()
self.agent = agent
self.name = name
async def invoke_async(self, task, **kwargs) -> MultiAgentResult:
# Execute function and create AgentResult
agent_result: AgentResult = self.agent(task if isinstance(task, str) else str(task))
result = random.choice([agent_result, Exception("OSError") ])
# Return wrapped in MultiAgentResult
return MultiAgentResult(
status = Status.FAILED if result is Exception else Status.COMPLETED,
results={
self.name: NodeResult(
result=result,
)
},
)
Basically a simple Agent Node but will occasionally (randomly) failed! But enough for us to get an idea of some of the components we have here!
First of all, we have the invoke_async function, the only abstract method we can and have to implement for this MultiAgentBase class.
For input argument, we have task which is either a plain str or list[ContentBlock]. And for the return value, we have MultiAgentResult with the following parameters.
- status: Execution status. Possible values: Status.Pending, Status.EXECUTING, Status.COMPLETED, Status.FAILED
- results : A map between the node’s name and NodeResult. The result of NodeResult can be an AgentResult, a MultiAgentResult, or an Exception. In addition, we can also optionally specify parameters such as execution_time, status for each node.
- accumulated_usage for inputTokens, outputTokens, and totalTokens used.
- accumulated_metrics for latency
- execution_count
- execution_time
We can then use it like following!
custom_agnet_node = RandomlyFailedAgentNode(agent=pokemon_helper, name="custom_agnet_node")
builder.add_node(custom_agnet_node, "custom_agnet_node")
Having a custom node type can be really useful in couple scenarios.
- The node doesn’t need to have LLM involved. For example, a node for data validation.
- Group multiple nodes as one for better organization, especially in the case of a more complex logic
Common Graph Flows
Here are couple common flow types that can be helpful! Of course, they are not mutually exclusive and we get to combine those the way we want to make the most out of it!
- Sequential or Process flow: visualize the step-by-step sequence of tasks in a process.
- Parallel Processing with Aggregation: What we had above! One input, one output, but multiple different routes to allow parallel processing.
- Branching: One input, but multiple branches with multiple outputs.
- Feedback Loop: A cyclic graph with revisit on specific nodes to allow revision.
If you would like a little more visualization, you can find it under the Common Graph Topologies provided by Strands. Also, I found that this 17 types of flowcharts and when to use them from Figma is also really useful!
Thank you for reading!
That’s it for this article!
What’s next?
Maybe Combining either Swarm or Graph with AgentCore Runtime so that we can have a somewhat self-automated agent that can run continuously for up to 8 hours!
Happy multi-agenting!
Strands Agents: Interesting Multi-Agent Pattern was originally published in Level Up Coding on Medium, where people are continuing the conversation by highlighting and responding to this story.
This content originally appeared on Level Up Coding – Medium and was authored by Itsuki