feat(multiagent): Swarm - PR feedback

Author: awsarron

src/strands/multiagent/base.py

@@ -82,11 +82,11 @@ class MultiAgentBase(ABC):
     """
 
     @abstractmethod
-    async def execute_async(self, task: str | list[ContentBlock], **kwargs: Any) -> MultiAgentResult:
-        """Execute task asynchronously."""
-        raise NotImplementedError("execute_async not implemented")
+    async def invoke_async(self, task: str | list[ContentBlock], **kwargs: Any) -> MultiAgentResult:
+        """Invoke asynchronously."""
+        raise NotImplementedError("invoke_async not implemented")
 
     @abstractmethod
     def __call__(self, task: str | list[ContentBlock], **kwargs: Any) -> MultiAgentResult:
-        """Execute task synchronously."""
+        """Invoke synchronously."""
         raise NotImplementedError("__call__ not implemented")

src/strands/multiagent/graph.py

@@ -140,6 +140,11 @@ def __init__(self) -> None:
 
     def add_node(self, executor: Agent | MultiAgentBase, node_id: str | None = None) -> GraphNode:
         """Add an Agent or MultiAgentBase instance as a node to the graph."""
+        # Check for duplicate node instances
+        seen_instances = {id(node.executor) for node in self.nodes.values()}
+        if id(executor) in seen_instances:
+            raise ValueError("Duplicate node instance detected. Each node must have a unique object instance.")
+
         # Auto-generate node_id if not provided
         if node_id is None:
             node_id = getattr(executor, "id", None) or getattr(executor, "name", None) or f"node_{len(self.nodes)}"
@@ -242,24 +247,27 @@ def __init__(self, nodes: dict[str, GraphNode], edges: set[GraphEdge], entry_poi
         """Initialize Graph."""
         super().__init__()
 
+        # Validate nodes for duplicate instances
+        self._validate_graph(nodes)
+
         self.nodes = nodes
         self.edges = edges
         self.entry_points = entry_points
         self.state = GraphState()
         self.tracer = get_tracer()
 
     def __call__(self, task: str | list[ContentBlock], **kwargs: Any) -> GraphResult:
-        """Execute task synchronously."""
+        """Invoke the graph synchronously."""
 
         def execute() -> GraphResult:
-            return asyncio.run(self.execute_async(task))
+            return asyncio.run(self.invoke_async(task))
 
         with ThreadPoolExecutor() as executor:
             future = executor.submit(execute)
             return future.result()
 
-    async def execute_async(self, task: str | list[ContentBlock], **kwargs: Any) -> GraphResult:
-        """Execute the graph asynchronously."""
+    async def invoke_async(self, task: str | list[ContentBlock], **kwargs: Any) -> GraphResult:
+        """Invoke the graph asynchronously."""
         logger.debug("task=<%s> | starting graph execution", task)
 
         # Initialize state
@@ -287,6 +295,15 @@ async def execute_async(self, task: str | list[ContentBlock], **kwargs: Any) ->
                 self.state.execution_time = round((time.time() - start_time) * 1000)
             return self._build_result()
 
+    def _validate_graph(self, nodes: dict[str, GraphNode]) -> None:
+        """Validate graph nodes for duplicate instances."""
+        # Check for duplicate node instances
+        seen_instances = set()
+        for node in nodes.values():
+            if id(node.executor) in seen_instances:
+                raise ValueError("Duplicate node instance detected. Each node must have a unique object instance.")
+            seen_instances.add(id(node.executor))
+
     async def _execute_graph(self) -> None:
         """Unified execution flow with conditional routing."""
         ready_nodes = list(self.entry_points)
@@ -349,7 +366,7 @@ async def _execute_node(self, node: GraphNode) -> None:
 
             # Execute based on node type and create unified NodeResult
             if isinstance(node.executor, MultiAgentBase):
-                multi_agent_result = await node.executor.execute_async(node_input)
+                multi_agent_result = await node.executor.invoke_async(node_input)
 
                 # Create NodeResult with MultiAgentResult directly
                 node_result = NodeResult(

src/strands/multiagent/swarm.py

@@ -13,6 +13,7 @@
 """
 
 import asyncio
+import copy
 import json
 import logging
 import time
@@ -21,8 +22,9 @@
 from typing import Any, Callable, Tuple, cast
 
 from ..agent import Agent, AgentResult
+from ..agent.state import AgentState
 from ..tools.decorator import tool
-from ..types.content import ContentBlock
+from ..types.content import ContentBlock, Messages
 from ..types.event_loop import Metrics, Usage
 from .base import MultiAgentBase, MultiAgentResult, NodeResult, Status
 
@@ -35,6 +37,14 @@ class SwarmNode:
 
     node_id: str
     executor: Agent
+    _initial_messages: Messages = field(default_factory=list, init=False)
+    _initial_state: AgentState = field(default_factory=AgentState, init=False)
+
+    def __post_init__(self) -> None:
+        """Capture initial executor state after initialization."""
+        # Deep copy the initial messages and state to preserve them
+        self._initial_messages = copy.deepcopy(self.executor.messages)
+        self._initial_state = AgentState(self.executor.state.get())
 
     def __hash__(self) -> int:
         """Return hash for SwarmNode based on node_id."""
@@ -54,6 +64,11 @@ def __repr__(self) -> str:
         """Return detailed representation of SwarmNode."""
         return f"SwarmNode(node_id='{self.node_id}')"
 
+    def reset_executor_state(self) -> None:
+        """Reset SwarmNode executor state to initial state when swarm was created."""
+        self.executor.messages = copy.deepcopy(self._initial_messages)
+        self.executor.state = AgentState(self._initial_state.get())
+
 
 @dataclass
 class SharedContext:
@@ -218,35 +233,19 @@ def __init__(
         self._setup_swarm(nodes)
         self._inject_swarm_tools()
 
-    def __call__(self, task: str | list[ContentBlock]) -> SwarmResult:
-        """Execute task synchronously.
-
-        Args:
-            task: The task to execute, either as a string or a list of ContentBlock objects
-                  for multi-modal content.
-
-        Returns:
-            SwarmResult containing execution results and metrics.
-        """
+    def __call__(self, task: str | list[ContentBlock], **kwargs: Any) -> SwarmResult:
+        """Invoke the swarm synchronously."""
 
         def execute() -> SwarmResult:
-            return asyncio.run(self.execute_async(task))
+            return asyncio.run(self.invoke_async(task))
 
         with ThreadPoolExecutor() as executor:
             future = executor.submit(execute)
             return future.result()
 
-    async def execute_async(self, task: str | list[ContentBlock]) -> SwarmResult:
-        """Execute the swarm asynchronously.
-
-        Args:
-            task: The task to execute, either as a string or a list of ContentBlock objects
-                  for multi-modal content.
-
-        Returns:
-            SwarmResult containing execution results and metrics.
-        """
-        logger.info("starting swarm execution")
+    async def invoke_async(self, task: str | list[ContentBlock], **kwargs: Any) -> SwarmResult:
+        """Invoke the swarm asynchronously."""
+        logger.debug("starting swarm execution")
 
         # Initialize swarm state with configuration
         initial_node = next(iter(self.nodes.values()))  # First SwarmNode
@@ -259,8 +258,8 @@ async def execute_async(self, task: str | list[ContentBlock]) -> SwarmResult:
 
         start_time = time.time()
         try:
-            logger.info("current_node=<%s> | starting swarm execution with node", self.state.current_node.node_id)
-            logger.info(
+            logger.debug("current_node=<%s> | starting swarm execution with node", self.state.current_node.node_id)
+            logger.debug(
                 "max_handoffs=<%d>, max_iterations=<%d>, timeout=<%s>s | swarm execution config",
                 self.max_handoffs,
                 self.max_iterations,
@@ -279,12 +278,15 @@ async def execute_async(self, task: str | list[ContentBlock]) -> SwarmResult:
 
     def _setup_swarm(self, nodes: list[Agent]) -> None:
         """Initialize swarm configuration."""
+        # Validate nodes before setup
+        self._validate_swarm(nodes)
+
         # Validate agents have names and create SwarmNode objects
         for i, node in enumerate(nodes):
             if not node.name:
                 node_id = f"node_{i}"
                 node.name = node_id
-                logger.info("node_id=<%s> | agent has no name, dynamically generating one", node_id)
+                logger.debug("node_id=<%s> | agent has no name, dynamically generating one", node_id)
 
             node_id = str(node.name)
 
@@ -295,7 +297,16 @@ def _setup_swarm(self, nodes: list[Agent]) -> None:
             self.nodes[node_id] = SwarmNode(node_id=node_id, executor=node)
 
         swarm_nodes = list(self.nodes.values())
-        logger.info("nodes=<%s> | initialized swarm with nodes", [node.node_id for node in swarm_nodes])
+        logger.debug("nodes=<%s> | initialized swarm with nodes", [node.node_id for node in swarm_nodes])
+
+    def _validate_swarm(self, nodes: list[Agent]) -> None:
+        """Validate swarm structure and nodes."""
+        # Check for duplicate object instances
+        seen_instances = set()
+        for node in nodes:
+            if id(node) in seen_instances:
+                raise ValueError("Duplicate node instance detected. Each node must have a unique object instance.")
+            seen_instances.add(id(node))
 
     def _inject_swarm_tools(self) -> None:
         """Add swarm coordination tools to each agent."""
@@ -324,7 +335,7 @@ def _inject_swarm_tools(self) -> None:
             # Use the agent's tool registry to process and register the tools
             node.executor.tool_registry.process_tools(swarm_tools)
 
-        logger.info(
+        logger.debug(
             "tool_count=<%d>, node_count=<%d> | injected coordination tools into agents",
             len(swarm_tools),
             len(self.nodes),
@@ -388,7 +399,7 @@ def _handle_handoff(self, target_node: SwarmNode, message: str, context: dict[st
         """Handle handoff to another agent."""
         # If task is already completed, don't allow further handoffs
         if self.state.completion_status != Status.EXECUTING:
-            logger.info(
+            logger.debug(
                 "task_status=<%s> | ignoring handoff request - task already completed",
                 self.state.completion_status,
             )
@@ -406,7 +417,7 @@ def _handle_handoff(self, target_node: SwarmNode, message: str, context: dict[st
             for key, value in context.items():
                 self.shared_context.add_context(previous_agent, key, value)
 
-        logger.info(
+        logger.debug(
             "from_node=<%s>, to_node=<%s> | handed off from agent to agent",
             previous_agent.node_id,
             target_node.node_id,
@@ -416,7 +427,7 @@ def _handle_completion(self) -> None:
         """Handle task completion."""
         self.state.completion_status = Status.COMPLETED
 
-        logger.info("swarm task completed")
+        logger.debug("swarm task completed")
 
     def _build_node_input(self, target_node: SwarmNode) -> str:
         """Build input text for a node based on shared context and handoffs.
@@ -499,7 +510,7 @@ async def _execute_swarm(self) -> None:
             while True:
                 if self.state.completion_status != Status.EXECUTING:
                     reason = f"Completion status is: {self.state.completion_status}"
-                    logger.info("reason=<%s> | stopping execution", reason)
+                    logger.debug("reason=<%s> | stopping execution", reason)
                     break
 
                 should_continue, reason = self.state.should_continue(
@@ -511,7 +522,7 @@ async def _execute_swarm(self) -> None:
                 )
                 if not should_continue:
                     self.state.completion_status = Status.FAILED
-                    logger.info("reason=<%s> | stopping execution", reason)
+                    logger.debug("reason=<%s> | stopping execution", reason)
                     break
 
                 # Get current node
@@ -521,13 +532,14 @@ async def _execute_swarm(self) -> None:
                     self.state.completion_status = Status.FAILED
                     break
 
-                logger.info(
+                logger.debug(
                     "current_node=<%s>, iteration=<%d> | executing node",
                     current_node.node_id,
                     len(self.state.node_history) + 1,
                 )
 
                 # Execute node with timeout protection
+                # TODO: Implement cancellation token to stop _execute_node from continuing
                 try:
                     await asyncio.wait_for(
                         self._execute_node(current_node, self.state.task),
@@ -536,11 +548,11 @@ async def _execute_swarm(self) -> None:
 
                     self.state.node_history.append(current_node)
 
-                    logger.info("node=<%s> | node execution completed", current_node.node_id)
+                    logger.debug("node=<%s> | node execution completed", current_node.node_id)
 
                     # Immediate check for completion after node execution
                     if self.state.completion_status != Status.EXECUTING:
-                        logger.info("status=<%s> | task completed with status", self.state.completion_status)  # type: ignore[unreachable]
+                        logger.debug("status=<%s> | task completed with status", self.state.completion_status)  # type: ignore[unreachable]
                         break
 
                 except asyncio.TimeoutError:
@@ -562,8 +574,8 @@ async def _execute_swarm(self) -> None:
             self.state.completion_status = Status.FAILED
 
         elapsed_time = time.time() - self.state.start_time
-        logger.info("status=<%s> | swarm execution completed", self.state.completion_status)
-        logger.info(
+        logger.debug("status=<%s> | swarm execution completed", self.state.completion_status)
+        logger.debug(
             "node_history_length=<%d>, time=<%s>s | metrics",
             len(self.state.node_history),
             f"{elapsed_time:.2f}",
@@ -588,7 +600,7 @@ async def _execute_node(self, node: SwarmNode, task: str | list[ContentBlock]) -
 
             # Execute node
             result = None
-            node.executor.messages = []  # Reset agent's messages to avoid polluting context
+            node.reset_executor_state()
             async for event in node.executor.stream_async(node_input):
                 if "result" in event:
                     result = cast(AgentResult, event["result"])

tests/strands/multiagent/test_base.py

@@ -138,7 +138,7 @@ class IncompleteMultiAgent(MultiAgentBase):
 
     # Test that complete implementations can be instantiated
     class CompleteMultiAgent(MultiAgentBase):
-        async def execute_async(self, task: str) -> MultiAgentResult:
+        async def invoke_async(self, task: str) -> MultiAgentResult:
             return MultiAgentResult(results={})
 
         def __call__(self, task: str) -> MultiAgentResult: