-
Notifications
You must be signed in to change notification settings - Fork 424
Description
Problem Statement
Customers have requested streaming support for multi-agent systems (Swarm and Graph), which currently buffer all responses until the entire execution completes before emitting any events. They want real-time feedback for long-running multi-agent workflows instead of waiting until everything finishes.
The current behavior:
- Single Agent: Streams events in real-time (text chunks, tool usage, reasoning, etc.)
- Multi-Agent (Swarm/Graph): No streaming - everything is buffered until completion
Customers have noted this inconsistency where switching from single agents to multi-agent patterns results in a different responsive streaming experience, as multi-agent systems requrie callback handlers to be set up to enable similar functionality.
Proposed Solution
Implement multi-agent streaming by emitting specialized events during multi-agent execution that mirror the single-agent streaming experience. This would involve:
New Multi-Agent Event Types
Extend the existing event system with multi-agent-specific events:
# Proposed event types in src/strands/types/_events.py
# (we should iterate on it)
class MultiAgentNodeStartEvent(TypedEvent):
"""Event emitted when a node (agent) begins execution in multi-agent context."""
def __init__(self, node_id: str, node_type: str, task: str | list[ContentBlock]):
super().__init__({
"multi_agent_node_start": True,
"node_id": node_id,
"node_type": node_type, # "agent", "swarm", "graph"
"task": task
})
class MultiAgentNodeStreamEvent(TypedEvent):
"""Event emitted during node execution - wraps single agent events."""
def __init__(self, node_id: str, agent_event: TypedEvent):
super().__init__({
"multi_agent_node_stream": True,
"node_id": node_id,
"agent_event": agent_event
})
class MultiAgentNodeCompleteEvent(TypedEvent):
"""Event emitted when a node completes execution."""
def __init__(self, node_id: str, result: NodeResult, execution_time: int):
super().__init__({
"multi_agent_node_complete": True,
"node_id": node_id,
"result": result,
"execution_time": execution_time
})
class MultiAgentHandoffEvent(TypedEvent):
"""Event emitted during agent handoffs in Swarm."""
def __init__(self, from_node: str, to_node: str, message: str):
super().__init__({
"multi_agent_handoff": True,
"from_node": from_node,
"to_node": to_node,
"handoff_message": message
})
Implementation in Swarm and Graph
Swarm Changes (src/strands/multiagent/swarm.py
):
- Emit
MultiAgentNodeStartEvent
before executing each agent - Forward agent streaming events wrapped in
MultiAgentNodeStreamEvent
- Emit
MultiAgentHandoffEvent
during agent handoffs - Emit
MultiAgentNodeCompleteEvent
after each agent execution
Graph Changes (src/strands/multiagent/graph.py
): - Emit
MultiAgentNodeStartEvent
for each graph node execution - Forward nested streaming events (from agents or nested multi-agent systems)
- Emit
MultiAgentNodeCompleteEvent
with dependency information
Event Emission During Execution
Multi-agent systems would emit events during execution, allowing users to access streaming through async iterators or by attaching callback handlers to individual agents within the multi-agent system.
Use Case
Long-running Multi-Agent Workflows: Users can see real-time progress as agents execute, hand off tasks, and generate responses, instead of waiting for the entire workflow to complete.
Debugging Multi-Agent Systems: Developers can observe which agent is currently executing, what tools they're using, and how data flows between agents.
User Interface Integration: Web applications can show live progress indicators, agent status, and streaming responses during complex multi-agent operations.
Example Workflow Visibility: See the example below of how developers can visualize the multiagent output stream.
[graph] Starting data_analyst: Analyze quarterly sales data
[data_analyst] Analyzing sales data from Q3...
[data_analyst] 🔧 Using tool: database_query
[data_analyst] Found 1,247 transactions totaling $89,432
[graph] Completed data_analyst (2,340ms)
[graph] Handoff: data_analyst → report_writer
[graph] Message: Data analysis complete, need formatted report
[graph] Starting report_writer: Create executive summary report
[report_writer] Creating executive summary...
[report_writer] 🔧 Using tool: chart_generator
[report_writer] # Q3 Sales Analysis Report...
[graph] Completed report_writer (1,890ms)
Alternatives Solutions
Multi-Agent Callback Handlers: Create specialized callback handlers for multi-agent systems - adds complexity and users can already attach handlers to individual agents.
Polling-based Progress: Add progress tracking APIs - requires additional complexity and doesn't provide real-time streaming experience.
Separate Streaming API: Create different methods for streaming vs non-streaming - breaks API consistency and adds confusion.
Additional Context
This feature would maintain backward compatibility since it only adds new event types. Existing callback handlers that don't handle multi-agent events would continue working unchanged.
The implementation should handle nested multi-agent systems (Graph containing Swarms) by properly forwarding and contextualizing events from nested executions.
This aligns with the existing streaming architecture and extends it naturally to multi-agent scenarios, providing a consistent user experience across all agent patterns.