feat: multiagent session interface

Author: JackYPCOnline

src/strands/experimental/multiagent_session/__init__.py

@@ -0,0 +1,26 @@
+"""Multi-agent session management for persistent execution.
+
+This package provides session persistence capabilities for multi-agent orchestrators,
+enabling resumable execution after interruptions or failures.
+"""
+
+from .multiagent_events import (
+    AfterGraphInvocationEvent,
+    AfterNodeInvocationEvent,
+    BeforeGraphInvocationEvent,
+    BeforeNodeInvocationEvent,
+    MultiAgentInitializationEvent,
+)
+from .multiagent_state import MultiAgentState, MultiAgentType
+from .multiagent_state_adapter import MultiAgentAdapter
+
+__all__ = [
+    "BeforeGraphInvocationEvent",
+    "AfterGraphInvocationEvent",
+    "MultiAgentInitializationEvent",
+    "BeforeNodeInvocationEvent",
+    "AfterNodeInvocationEvent",
+    "MultiAgentState",
+    "MultiAgentAdapter",
+    "MultiAgentType",
+]

src/strands/experimental/multiagent_session/multiagent_events.py

@@ -0,0 +1,87 @@
+"""Multi-agent execution lifecycle events for hook system integration.
+
+This module defines event classes that are triggered at key points during
+multi-agent orchestrator execution, enabling hooks to respond to lifecycle
+events for purposes like persistence, monitoring, and debugging.
+
+Event Types:
+- Initialization: When orchestrator starts up
+- Before/After Graph: Start/end of overall execution
+- Before/After Node: Start/end of individual node execution
+"""
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from ...hooks.registry import HookEvent
+from .multiagent_state import MultiAgentState
+
+if TYPE_CHECKING:
+    from ...multiagent.base import MultiAgentBase
+
+
+@dataclass
+class MultiAgentInitializationEvent(HookEvent):
+    """Event triggered when multi-agent orchestrator initializes.
+
+    Attributes:
+        orchestrator: The multi-agent orchestrator instance
+        state: Current state of the orchestrator
+    """
+
+    orchestrator: "MultiAgentBase"
+    state: MultiAgentState
+
+
+@dataclass
+class BeforeGraphInvocationEvent(HookEvent):
+    """Event triggered before orchestrator execution begins.
+
+    Attributes:
+        orchestrator: The multi-agent orchestrator instance
+        state: Current state before execution starts
+    """
+
+    orchestrator: "MultiAgentBase"
+    state: MultiAgentState
+
+
+@dataclass
+class BeforeNodeInvocationEvent(HookEvent):
+    """Event triggered before individual node execution.
+
+    Attributes:
+        orchestrator: The multi-agent orchestrator instance
+        next_node_to_execute: ID of the node about to be executed
+    """
+
+    orchestrator: "MultiAgentBase"
+    next_node_to_execute: str
+
+
+@dataclass
+class AfterNodeInvocationEvent(HookEvent):
+    """Event triggered after individual node execution completes.
+
+    Attributes:
+        orchestrator: The multi-agent orchestrator instance
+        executed_node: ID of the node that just completed execution
+        state: Updated state after node execution
+    """
+
+    orchestrator: "MultiAgentBase"
+    executed_node: str
+    state: MultiAgentState
+
+
+@dataclass
+class AfterGraphInvocationEvent(HookEvent):
+    """Event triggered after orchestrator execution completes.
+
+    Attributes:
+        orchestrator: The multi-agent orchestrator instance
+        state: Final state after execution completes
+    """
+
+    orchestrator: "MultiAgentBase"
+    state: MultiAgentState

src/strands/experimental/multiagent_session/multiagent_state.py

@@ -0,0 +1,110 @@
+"""Multi-agent state data structures for session persistence.
+
+This module defines the core data structures used to represent the state
+of multi-agent orchestrators in a serializable format for session persistence.
+
+Key Components:
+- MultiAgentType: Enum for orchestrator types (Graph/Swarm)
+- MultiAgentState: Serializable state container with conversion methods
+"""
+
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set
+
+from ...types.content import ContentBlock
+
+if TYPE_CHECKING:
+    from ...multiagent.base import Status
+
+
+# TODO: Move to Base after experimental
+class MultiAgentType(Enum):
+    """Enumeration of supported multi-agent orchestrator types.
+
+    Attributes:
+        SWARM: Collaborative agent swarm orchestrator
+        GRAPH: Directed graph-based agent orchestrator
+    """
+
+    SWARM = "swarm"
+    GRAPH = "graph"
+
+
+@dataclass
+class MultiAgentState:
+    """Serializable state container for multi-agent orchestrators.
+
+    This class represents the complete execution state of a multi-agent
+    orchestrator (Graph or Swarm) in a format suitable for persistence
+    and restoration across sessions.
+
+    Attributes:
+        completed_nodes: Set of node IDs that have completed execution
+        node_results: Dictionary mapping node IDs to their execution results
+        status: Current execution status of the orchestrator
+        next_node_to_execute: List of node IDs ready for execution
+        current_task: The original task being executed
+        execution_order: Ordered list of executed node IDs
+        error_message: Optional error message if execution failed
+        type: Type of orchestrator (Graph or Swarm)
+        context: Additional context data (primarily for Swarm)
+    """
+
+    # Mutual
+    completed_nodes: Set[str] = field(default_factory=set)
+    node_results: Dict[str, Any] = field(default_factory=dict)
+    status: "Status" = "pending"
+    next_node_to_execute: Optional[List[str]] = None
+    current_task: Optional[str | List[ContentBlock]] = None
+    execution_order: list[str] = field(default_factory=list)
+    error_message: Optional[str] = None
+    type: Optional[MultiAgentType] = field(default=MultiAgentType.GRAPH)
+    # Swarm
+    context: Optional[dict] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        """Convert MultiAgentState to JSON-serializable dictionary.
+
+        Returns:
+            Dictionary representation suitable for JSON serialization
+        """
+
+        def _serialize(v: Any) -> Any:
+            if isinstance(v, (str, int, float, bool)) or v is None:
+                return v
+            if isinstance(v, set):
+                return list(v)
+            if isinstance(v, dict):
+                return {str(k): _serialize(val) for k, val in v.items()}
+            if isinstance(v, (list, tuple)):
+                return [_serialize(x) for x in v]
+            if hasattr(v, "to_dict"):
+                return v.to_dict()
+            # last resort: stringize anything non-serializable (locks, objects, etc.)
+            return str(v)
+
+        return {
+            "status": self.status,
+            "completed_nodes": list(self.completed_nodes),
+            "next_node_to_execute": list(self.next_node_to_execute) if self.next_node_to_execute else [],
+            "node_results": _serialize(self.node_results),
+            "current_task": self.current_task,
+            "error_message": self.error_message,
+            "execution_order": self.execution_order,
+            "type": self.type,
+            "context": _serialize(self.context),
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict):
+        """Create MultiAgentState from dictionary data.
+
+        Args:
+            data: Dictionary containing state data
+
+        Returns:
+            MultiAgentState instance
+        """
+        data["completed_nodes"] = set(data.get("completed_nodes", []))
+        return cls(**data)