@ -5,6 +5,7 @@ from typing import Callable, List, Optional, Union
from swarms . structs . agent import Agent
from swarms . structs . base_swarm import BaseSwarm
from swarms . structs . conversation import Conversation
from swarms . structs . swarm_id import swarm_id
from swarms . utils . formatter import formatter
from swarms . utils . get_cpu_cores import get_cpu_cores
from swarms . utils . history_output_formatter import (
@ -16,131 +17,14 @@ logger = initialize_logger(log_folder="concurrent_workflow")
class ConcurrentWorkflow ( BaseSwarm ) :
"""
A production - grade concurrent workflow orchestrator that executes multiple agents simultaneously .
ConcurrentWorkflow is designed for high - performance multi - agent orchestration with advanced features
including real - time monitoring , error handling , caching , and flexible output formatting . It ' s ideal
for scenarios where multiple agents need to process the same task independently and their results
need to be aggregated .
Key Features :
- Concurrent execution using ThreadPoolExecutor for optimal performance
- Real - time dashboard monitoring with status updates
- Comprehensive error handling and recovery
- Flexible output formatting options
- Automatic prompt engineering capabilities
- Conversation history management
- Metadata persistence and auto - saving
- Support for both single and batch processing
- Image input support for multimodal agents
Use Cases :
- Multi - perspective analysis ( financial , legal , technical reviews )
- Consensus building and voting systems
- Parallel data processing and analysis
- A / B testing with different agent configurations
- Redundancy and reliability improvements
Args :
name ( str , optional ) : Unique identifier for the workflow instance .
Defaults to " ConcurrentWorkflow " .
description ( str , optional ) : Human - readable description of the workflow ' s purpose.
Defaults to " Execution of multiple agents concurrently " .
agents ( List [ Union [ Agent , Callable ] ] , optional ) : List of Agent instances or callable objects
to execute concurrently . Each agent should implement a ` run ` method .
Defaults to empty list .
metadata_output_path ( str , optional ) : File path for saving execution metadata and results .
Supports JSON format . Defaults to " agent_metadata.json " .
auto_save ( bool , optional ) : Whether to automatically save conversation history and metadata
after each run . Defaults to True .
output_type ( str , optional ) : Format for aggregating agent outputs . Options include :
- " dict-all-except-first " : Dictionary with all agent outputs except the first
- " dict " : Dictionary with all agent outputs
- " str " : Concatenated string of all outputs
- " list " : List of individual agent outputs
Defaults to " dict-all-except-first " .
max_loops ( int , optional ) : Maximum number of execution loops for each agent .
Defaults to 1.
auto_generate_prompts ( bool , optional ) : Enable automatic prompt engineering for all agents .
When True , agents will enhance their prompts automatically . Defaults to False .
show_dashboard ( bool , optional ) : Enable real - time dashboard display showing agent status ,
progress , and outputs . Useful for monitoring and debugging . Defaults to False .
* args : Additional positional arguments passed to the BaseSwarm parent class .
* * kwargs : Additional keyword arguments passed to the BaseSwarm parent class .
Raises :
ValueError : If agents list is empty or None .
ValueError : If description is empty or None .
TypeError : If agents list contains non - Agent , non - callable objects .
Attributes :
name ( str ) : The workflow instance name .
description ( str ) : The workflow description .
agents ( List [ Union [ Agent , Callable ] ] ) : List of agents to execute .
metadata_output_path ( str ) : Path for metadata output file .
auto_save ( bool ) : Auto - save flag for metadata persistence .
output_type ( str ) : Output aggregation format .
max_loops ( int ) : Maximum execution loops per agent .
auto_generate_prompts ( bool ) : Auto prompt engineering flag .
show_dashboard ( bool ) : Dashboard display flag .
agent_statuses ( dict ) : Real - time status tracking for each agent .
conversation ( Conversation ) : Conversation history manager .
Example :
Basic usage with multiple agents :
>> > from swarms import Agent , ConcurrentWorkflow
>> >
>> > # Create specialized agents
>> > financial_agent = Agent (
. . . agent_name = " Financial-Analyst " ,
. . . system_prompt = " You are a financial analysis expert... " ,
. . . model_name = " gpt-4 "
. . . )
>> > legal_agent = Agent (
. . . agent_name = " Legal-Advisor " ,
. . . system_prompt = " You are a legal expert... " ,
. . . model_name = " gpt-4 "
. . . )
>> >
>> > # Create workflow
>> > workflow = ConcurrentWorkflow (
. . . name = " Multi-Expert-Analysis " ,
. . . agents = [ financial_agent , legal_agent ] ,
. . . show_dashboard = True ,
. . . auto_save = True
. . . )
>> >
>> > # Execute analysis
>> > task = " Analyze the risks of investing in cryptocurrency "
>> > results = workflow . run ( task )
>> > print ( f " Analysis complete with { len ( results ) } perspectives " )
Batch processing example :
>> > tasks = [
. . . " Analyze Q1 financial performance " ,
. . . " Review Q2 market trends " ,
. . . " Forecast Q3 projections "
. . . ]
>> > batch_results = workflow . batch_run ( tasks )
>> > print ( f " Processed { len ( batch_results ) } quarterly reports " )
Note :
- Agents are executed using ThreadPoolExecutor with 95 % of available CPU cores
- Each agent runs independently and cannot communicate with others during execution
- The workflow maintains conversation history across all runs for context
- Dashboard mode disables individual agent printing to prevent output conflicts
- Error handling ensures partial results are available even if some agents fail
"""
""" Concurrent workflow for running multiple agents simultaneously. """
def __init__ (
self ,
id : str = swarm_id ( ) ,
name : str = " ConcurrentWorkflow " ,
description : str = " Execution of multiple agents concurrently " ,
agents : List [ Union [ Agent , Callable ] ] = None ,
metadata_output_path : str = " agent_metadata.json " ,
auto_save : bool = True ,
output_type : str = " dict-all-except-first " ,
max_loops : int = 1 ,
@ -149,17 +33,6 @@ class ConcurrentWorkflow(BaseSwarm):
* args ,
* * kwargs ,
) :
"""
Initialize the ConcurrentWorkflow with configuration parameters .
Performs initialization , validation , and setup of internal components including
conversation management , agent status tracking , and dashboard configuration .
Note :
The constructor automatically performs reliability checks and configures
agents for dashboard mode if enabled . Agent print outputs are disabled
when dashboard mode is active to prevent display conflicts .
"""
super ( ) . __init__ (
name = name ,
description = description ,
@ -170,7 +43,9 @@ class ConcurrentWorkflow(BaseSwarm):
self . name = name
self . description = description
self . agents = agents
self . metadata_output_path = metadata_output_path
self . metadata_output_path = (
f " concurrent_workflow_name_ { name } _id_ { id } .json "
)
self . auto_save = auto_save
self . max_loops = max_loops
self . auto_generate_prompts = auto_generate_prompts
@ -188,58 +63,14 @@ class ConcurrentWorkflow(BaseSwarm):
self . agents = self . fix_agents ( )
def fix_agents ( self ) :
"""
Configure agents for dashboard mode by disabling individual print outputs .
When dashboard mode is enabled , individual agent print outputs can interfere
with the dashboard display . This method disables print_on for all agents
to ensure clean dashboard rendering .
Returns :
List [ Agent ] : The modified list of agents with print_on disabled .
Note :
This method only modifies agents when show_dashboard is True .
Agent functionality is not affected , only their output display behavior .
Example :
>> > workflow = ConcurrentWorkflow ( show_dashboard = True , agents = [ agent1 , agent2 ] )
>> > # Agents automatically configured for dashboard mode
>> > all ( not agent . print_on for agent in workflow . agents )
True
"""
""" Configure agents for dashboard mode. """
if self . show_dashboard is True :
for agent in self . agents :
agent . print_on = False
return self . agents
def reliability_check ( self ) :
"""
Perform comprehensive validation of workflow configuration and agents .
Validates that the workflow is properly configured with valid agents and
provides warnings for suboptimal configurations . This method is called
automatically during initialization .
Validates :
- Agents list is not None or empty
- At least one agent is provided
- Warns if only one agent is provided ( suboptimal for concurrent execution )
Raises :
ValueError : If agents list is None or empty .
Logs :
Warning : If only one agent is provided ( concurrent execution not beneficial ) .
Error : If validation fails with detailed error information .
Example :
>> > workflow = ConcurrentWorkflow ( agents = [ ] )
ValueError : ConcurrentWorkflow : No agents provided
>> > workflow = ConcurrentWorkflow ( agents = [ single_agent ] )
# Logs warning about single agent usage
"""
""" Validate workflow configuration. """
try :
if self . agents is None :
raise ValueError (
@ -253,7 +84,7 @@ class ConcurrentWorkflow(BaseSwarm):
if len ( self . agents ) == 1 :
logger . warning (
" ConcurrentWorkflow: Only one agent provided. With ConcurrentWorkflow, you should use at least 2+ agents. "
" ConcurrentWorkflow: Only one agent provided. "
)
except Exception as e :
logger . error (
@ -262,33 +93,7 @@ class ConcurrentWorkflow(BaseSwarm):
raise
def activate_auto_prompt_engineering ( self ) :
"""
Enable automatic prompt engineering for all agents in the workflow .
When activated , each agent will automatically enhance and optimize their
system prompts based on the task context and their previous interactions .
This can improve response quality but may increase execution time .
Side Effects :
- Sets auto_generate_prompt = True for all agents in the workflow
- Affects subsequent agent . run ( ) calls , not retroactively
- May increase latency due to prompt optimization overhead
Note :
This method can be called at any time , but changes only affect
future agent executions . Already running agents are not affected .
Example :
>> > workflow = ConcurrentWorkflow ( agents = [ agent1 , agent2 ] )
>> > workflow . activate_auto_prompt_engineering ( )
>> > # All agents will now auto-generate optimized prompts
>> > result = workflow . run ( " Complex analysis task " )
>> > # Agents used enhanced prompts for better performance
See Also :
- Agent . auto_generate_prompt : Individual agent prompt engineering
- auto_generate_prompts : Constructor parameter for default behavior
"""
""" Enable automatic prompt engineering. """
if self . auto_generate_prompts is True :
for agent in self . agents :
agent . auto_generate_prompt = True
@ -297,48 +102,8 @@ class ConcurrentWorkflow(BaseSwarm):
self ,
title : str = " ConcurrentWorkflow Dashboard " ,
is_final : bool = False ,
) - > None :
"""
Display a real - time dashboard showing the current status of all agents .
Renders a formatted dashboard with agent names , execution status , and
output previews . The dashboard is updated in real - time during workflow
execution to provide visibility into agent progress and results .
Args :
title ( str , optional ) : The dashboard title to display at the top .
Defaults to " 🤖 ConcurrentWorkflow Dashboard " .
is_final ( bool , optional ) : Whether this is the final dashboard display
after all agents have completed . Changes formatting and styling .
Defaults to False .
Side Effects :
- Prints formatted dashboard to console
- Updates display in real - time during execution
- May clear previous dashboard content for clean updates
Note :
This method is automatically called during workflow execution when
show_dashboard = True . Manual calls are supported for custom monitoring .
Dashboard Status Values :
- " pending " : Agent queued but not yet started
- " running " : Agent currently executing task
- " completed " : Agent finished successfully
- " error " : Agent execution failed with error
Example :
>> > workflow = ConcurrentWorkflow ( agents = [ agent1 , agent2 ] , show_dashboard = True )
>> > workflow . display_agent_dashboard ( " Custom Dashboard Title " )
# Displays:
# 🤖 Custom Dashboard Title
# ┌─────────────────┬─────────┬──────────────────┐
# │ Agent Name │ Status │ Output Preview │
# ├─────────────────┼─────────┼──────────────────┤
# │ Financial-Agent │ running │ Analyzing data...│
# │ Legal-Agent │ pending │ │
# └─────────────────┴─────────┴──────────────────┘
"""
) :
""" Display real-time dashboard. """
agents_data = [
{
" name " : agent . agent_name ,
@ -358,69 +123,11 @@ class ConcurrentWorkflow(BaseSwarm):
task : str ,
img : Optional [ str ] = None ,
imgs : Optional [ List [ str ] ] = None ,
streaming_callback : Optional [
Callable [ [ str , str , bool ] , None ]
] = None ,
) :
"""
Execute all agents concurrently with real - time dashboard monitoring .
This method provides the same concurrent execution as _run ( ) but with
enhanced real - time monitoring through a visual dashboard . Agent status
updates are displayed in real - time , showing progress , completion , and
any errors that occur during execution .
Args :
task ( str ) : The task description or prompt to be executed by all agents .
This will be passed to each agent ' s run() method.
img ( Optional [ str ] , optional ) : Path to a single image file for agents
that support multimodal input . Defaults to None .
imgs ( Optional [ List [ str ] ] , optional ) : List of image file paths for
agents that support multiple image inputs . Defaults to None .
Returns :
Any : Formatted output based on the configured output_type setting .
The return format depends on the output_type parameter set during
workflow initialization .
Raises :
Exception : Re - raises any exceptions from agent execution after
updating the dashboard with error status .
Side Effects :
- Displays initial dashboard showing all agents as " pending "
- Updates dashboard in real - time as agents start , run , and complete
- Displays final dashboard with all results when execution completes
- Adds all task inputs and agent outputs to conversation history
- Automatically cleans up dashboard display resources
Dashboard Flow :
1. Initial dashboard shows all agents as " pending "
2. As agents start , status updates to " running "
3. As agents complete , status updates to " completed " with output preview
4. Final dashboard shows complete results summary
5. Dashboard resources are cleaned up automatically
Performance :
- Uses 95 % of available CPU cores for optimal concurrency
- ThreadPoolExecutor manages thread lifecycle automatically
- Real - time updates have minimal performance overhead
Example :
>> > workflow = ConcurrentWorkflow (
. . . agents = [ financial_agent , legal_agent ] ,
. . . show_dashboard = True
. . . )
>> > result = workflow . run_with_dashboard (
. . . task = " Analyze the merger proposal " ,
. . . img = " company_financials.png "
. . . )
# Dashboard shows real-time progress:
# Agent-1: pending -> running -> completed
# Agent-2: pending -> running -> completed
>> > print ( " Analysis complete: " , result )
Note :
This method is automatically called when show_dashboard = True .
For headless execution without dashboard , use _run ( ) directly .
"""
""" Execute agents with dashboard monitoring. """
try :
self . conversation . add ( role = " User " , content = task )
@ -431,61 +138,61 @@ class ConcurrentWorkflow(BaseSwarm):
" output " : " " ,
}
# Display initial dashboard if enabled
if self . show_dashboard :
self . display_agent_dashboard ( )
# Use 95% of available CPU cores for optimal performance
max_workers = int ( get_cpu_cores ( ) * 0.95 )
# Create a list to store all futures and their results
futures = [ ]
results = [ ]
def run_agent_with_status ( agent , task , img , imgs ) :
try :
# Update status to running
self . agent_statuses [ agent . agent_name ] [
" status "
] = " running "
if self . show_dashboard :
self . display_agent_dashboard ( )
# Create a streaming callback for this agent with throttling
last_update_time = [
0
] # Use list to allow modification in nested function
update_interval = 0.1 # Update dashboard every 100ms for smooth streaming
def streaming_callback ( chunk : str ) :
""" Update dashboard with streaming content """
if self . show_dashboard :
# Append the chunk to the agent's current output
current_output = self . agent_statuses [
agent . agent_name
] [ " output " ]
self . agent_statuses [ agent . agent_name ] [
" output "
] = ( current_output + chunk )
# Throttle dashboard updates for better performance
current_time = time . time ( )
last_update_time = [ 0 ]
update_interval = 0.1
def agent_streaming_callback ( chunk : str ) :
try :
if self . show_dashboard and chunk :
current_output = self . agent_statuses [
agent . agent_name
] [ " output " ]
self . agent_statuses [ agent . agent_name ] [
" output "
] = ( current_output + chunk )
current_time = time . time ( )
if (
current_time - last_update_time [ 0 ]
> = update_interval
) :
self . display_agent_dashboard ( )
last_update_time [ 0 ] = current_time
if (
current_time - last_update_time [ 0 ]
> = update_interval
streaming_callback
and chunk is not None
) :
self . display_agent_dashboard ( )
last_update_time [ 0 ] = current_time
streaming_callback (
agent . agent_name , chunk , False
)
except Exception as callback_error :
logger . warning (
f " Dashboard streaming callback failed for { agent . agent_name } : { str ( callback_error ) } "
)
# Run the agent with streaming callback
output = agent . run (
task = task ,
img = img ,
imgs = imgs ,
streaming_callback = ,
streaming_callback = agent_ streaming_callback,
)
# Update status to completed
self . agent_statuses [ agent . agent_name ] [
" status "
] = " completed "
@ -495,9 +202,11 @@ class ConcurrentWorkflow(BaseSwarm):
if self . show_dashboard :
self . display_agent_dashboard ( )
if streaming_callback :
streaming_callback ( agent . agent_name , " " , True )
return output
except Exception as e :
# Update status to error
self . agent_statuses [ agent . agent_name ] [
" status "
] = " error "
@ -506,24 +215,25 @@ class ConcurrentWorkflow(BaseSwarm):
] = f " Error: { str ( e ) } "
if self . show_dashboard :
self . display_agent_dashboard ( )
if streaming_callback :
streaming_callback (
agent . agent_name , f " Error: { str ( e ) } " , True
)
raise
# Run agents concurrently using ThreadPoolExecutor
with concurrent . futures . ThreadPoolExecutor (
max_workers = max_workers
) as executor :
# Submit all agent tasks
futures = [
executor . submit (
run_agent_with_status , agent , task , img , imgs
)
for agent in self . agents
]
# Wait for all futures to complete
concurrent . futures . wait ( futures )
# Process results in order of completion
for future , agent in zip ( futures , self . agents ) :
try :
output = future . result ( )
@ -536,11 +246,9 @@ class ConcurrentWorkflow(BaseSwarm):
( agent . agent_name , f " Error: { str ( e ) } " )
)
# Add all results to conversation
for agent_name , output in results :
self . conversation . add ( role = agent_name , content = output )
# Display final dashboard if enabled
if self . show_dashboard :
self . display_agent_dashboard (
" Final ConcurrentWorkflow Dashboard " ,
@ -548,11 +256,9 @@ class ConcurrentWorkflow(BaseSwarm):
)
return history_output_formatter (
conversation = self . conversation ,
type = self . output_type ,
conversation = self . conversation , type = self . output_type
)
finally :
# Always clean up the dashboard display
if self . show_dashboard :
formatter . stop_dashboard ( )
@ -561,316 +267,160 @@ class ConcurrentWorkflow(BaseSwarm):
task : str ,
img : Optional [ str ] = None ,
imgs : Optional [ List [ str ] ] = None ,
streaming_callback : Optional [
Callable [ [ str , str , bool ] , None ]
] = None ,
) :
"""
Execute all agents concurrently without dashboard monitoring ( headless mode ) .
This is the core execution method that runs all agents simultaneously using
ThreadPoolExecutor for optimal performance . Results are collected and
formatted according to the configured output_type .
Args :
task ( str ) : The task description or prompt to be executed by all agents .
Each agent receives the same task input for independent processing .
img ( Optional [ str ] , optional ) : Path to a single image file for multimodal
agents . The same image is provided to all agents . Defaults to None .
imgs ( Optional [ List [ str ] ] , optional ) : List of image file paths for agents
that support multiple image inputs . All agents receive the same image list .
Defaults to None .
Returns :
Any : Formatted output according to the output_type configuration :
- " dict-all-except-first " : Dict with all agent outputs except first
- " dict " : Dict with all agent outputs keyed by agent name
- " str " : Concatenated string of all agent outputs
- " list " : List of individual agent outputs in completion order
Side Effects :
- Adds the user task to conversation history
- Adds each agent ' s output to conversation history upon completion
- No visual output or dashboard updates ( headless execution )
Performance Characteristics :
- Uses ThreadPoolExecutor with 95 % of available CPU cores
- Agents execute truly concurrently , not sequentially
- Results are collected in completion order , not submission order
- Memory efficient for large numbers of agents
Error Handling :
- Individual agent failures don ' t stop other agents
- Failed agents have their exceptions logged but execution continues
- Partial results are still returned for successful agents
Thread Safety :
- Conversation object handles concurrent access safely
- Agent status is not tracked in this method ( dashboard - free )
- Each agent runs in isolation without shared state
Example :
>> > workflow = ConcurrentWorkflow (
. . . agents = [ agent1 , agent2 , agent3 ] ,
. . . output_type = " dict "
. . . )
>> > result = workflow . _run (
. . . task = " Analyze market trends for Q4 " ,
. . . img = " market_chart.png "
. . . )
>> > print ( f " Received { len ( result ) } agent analyses " )
>> > # Result format: {"agent1": "analysis1", "agent2": "analysis2", ...}
Note :
This method is called automatically by run ( ) when show_dashboard = False .
For monitoring and real - time updates , use run_with_dashboard ( ) instead .
"""
""" Execute agents concurrently without dashboard. """
self . conversation . add ( role = " User " , content = task )
# Use 95% of available CPU cores for optimal performance
max_workers = int ( get_cpu_cores ( ) * 0.95 )
# Run agents concurrently using ThreadPoolExecutor
with concurrent . futures . ThreadPoolExecutor (
max_workers = max_workers
) as executor :
# Submit all agent tasks and store with their index
future_to_agent = {
executor . submit (
agent . run , task = task , img = img , imgs = imgs
self . _run_agent_with_streaming ,
agent ,
task ,
img ,
imgs ,
streaming_callback ,
) : agent
for agent in self . agents
}
# Collect results and add to conversation in completion order
for future in concurrent . futures . as_completed (
future_to_agent
) :
agent = future_to_agent [ future ]
output = future . result ( )
self . conversation . add ( role = agent . name , content = output )
self . conversation . add (
role = agent . agent_name , content = output
)
return history_output_formatter (
conversation = self . conversation ,
type = self . output_type ,
conversation = self . conversation , type = self . output_type
)
def _run_agent_with_streaming (
self ,
agent : Union [ Agent , Callable ] ,
task : str ,
img : Optional [ str ] = None ,
imgs : Optional [ List [ str ] ] = None ,
streaming_callback : Optional [
Callable [ [ str , str , bool ] , None ]
] = None ,
) :
""" Run single agent with streaming support. """
if streaming_callback is None :
return agent . run ( task = task , img = img , imgs = imgs )
def agent_streaming_callback ( chunk : str ) :
try :
# Safely call the streaming callback
if streaming_callback and chunk is not None :
streaming_callback ( agent . agent_name , chunk , False )
except Exception as callback_error :
logger . warning (
f " Streaming callback failed for { agent . agent_name } : { str ( callback_error ) } "
)
try :
output = agent . run (
task = task ,
img = img ,
imgs = imgs ,
streaming_callback = agent_streaming_callback ,
)
# Ensure completion callback is called even if there were issues
try :
streaming_callback ( agent . agent_name , " " , True )
except Exception as callback_error :
logger . warning (
f " Completion callback failed for { agent . agent_name } : { str ( callback_error ) } "
)
return output
except Exception as e :
error_msg = f " Agent { agent . agent_name } failed: { str ( e ) } "
logger . error ( error_msg )
# Try to send error through callback
try :
streaming_callback (
agent . agent_name , f " Error: { str ( e ) } " , True
)
except Exception as callback_error :
logger . warning (
f " Error callback failed for { agent . agent_name } : { str ( callback_error ) } "
)
raise
def cleanup ( self ) :
""" Clean up resources and connections. """
try :
# Reset agent statuses
for agent in self . agents :
if hasattr ( agent , " cleanup " ) :
try :
agent . cleanup ( )
except Exception as e :
logger . warning (
f " Failed to cleanup agent { agent . agent_name } : { str ( e ) } "
)
# Clear conversation if needed
if hasattr ( self , " conversation " ) :
# Keep the conversation for result formatting but reset for next run
pass
except Exception as e :
logger . error ( f " Cleanup failed: { str ( e ) } " )
def run (
self ,
task : str ,
img : Optional [ str ] = None ,
imgs : Optional [ List [ str ] ] = None ,
streaming_callback : Optional [
Callable [ [ str , str , bool ] , None ]
] = None ,
) :
"""
Execute all agents concurrently on the given task with optional dashboard monitoring .
This is the main entry point for workflow execution . Automatically selects
between dashboard - enabled execution ( run_with_dashboard ) or headless execution
( _run ) based on the show_dashboard configuration .
Args :
task ( str ) : The task description , prompt , or instruction to be executed
by all agents concurrently . Each agent processes the same task
independently .
img ( Optional [ str ] , optional ) : Path to a single image file for agents
that support multimodal ( text + image ) input . Defaults to None .
imgs ( Optional [ List [ str ] ] , optional ) : List of image file paths for
agents that support multiple image inputs . Defaults to None .
Returns :
Any : Aggregated results from all agents formatted according to output_type :
- " dict-all-except-first " : Dictionary excluding first agent ' s output
- " dict " : Complete dictionary with all agent outputs
- " str " : Concatenated string of all outputs
- " list " : List of individual agent outputs
Execution Modes :
- Dashboard Mode ( show_dashboard = True ) : Provides real - time visual monitoring
with status updates , progress tracking , and error visibility
- Headless Mode ( show_dashboard = False ) : Silent execution with no visual output ,
optimized for performance and automation scenarios
Concurrent Execution :
- All agents run simultaneously using ThreadPoolExecutor
- Utilizes 95 % of available CPU cores for optimal performance
- Thread - safe conversation history management
- Independent agent execution without inter - agent communication
Error Resilience :
- Individual agent failures don ' t halt the entire workflow
- Partial results are returned for successful agents
- Comprehensive error logging for debugging
- Graceful degradation under failure conditions
Example :
Basic concurrent execution :
>> > workflow = ConcurrentWorkflow ( agents = [ analyst , reviewer , summarizer ] )
>> > result = workflow . run ( " Evaluate the new product proposal " )
>> > print ( f " Received insights from { len ( workflow . agents ) } experts " )
With dashboard monitoring :
>> > workflow = ConcurrentWorkflow (
. . . agents = [ financial_agent , legal_agent ] ,
. . . show_dashboard = True
. . . )
>> > result = workflow . run ( " Review merger agreement " )
# Real-time dashboard shows progress and completion status
Multimodal analysis :
>> > result = workflow . run (
. . . task = " Analyze this chart and provide insights " ,
. . . img = " quarterly_results.png "
. . . )
Performance Tips :
- Use 2 + agents to benefit from concurrent execution
- Dashboard mode adds minimal overhead for monitoring
- Larger agent counts scale well with available CPU cores
- Consider batch_run ( ) for multiple related tasks
See Also :
- batch_run ( ) : For processing multiple tasks sequentially
- run_with_dashboard ( ) : For direct dashboard execution
- _run ( ) : For direct headless execution
"""
if self . show_dashboard :
return self . run_with_dashboard ( task , img , imgs )
else :
return self . _run ( task , img , imgs )
""" Execute all agents concurrently. """
try :
if self . show_dashboard :
result = self . run_with_dashboard (
task , img , imgs , streaming_callback
)
else :
result = self . _run (
task , img , imgs , streaming_callback
)
return result
finally :
# Always cleanup resources
self . cleanup ( )
def batch_run (
self ,
tasks : List [ str ] ,
imgs : Optional [ List [ str ] ] = None ,
streaming_callback : Optional [
Callable [ [ str , str , bool ] , None ]
] = None ,
) :
"""
Execute the workflow on multiple tasks sequentially with concurrent agent processing .
This method processes a list of tasks one by one , where each task is executed
by all agents concurrently . This is ideal for batch processing scenarios where
you have multiple related tasks that need the same multi - agent analysis .
Args :
tasks ( List [ str ] ) : List of task descriptions , prompts , or instructions .
Each task will be processed by all agents concurrently before
moving to the next task . Tasks are processed sequentially .
imgs ( Optional [ List [ str ] ] , optional ) : List of image file paths corresponding
to each task . If provided , imgs [ i ] will be used for tasks [ i ] .
If fewer images than tasks are provided , remaining tasks will
execute without images . Defaults to None .
Returns :
List [ Any ] : List of results , one for each task . Each result is formatted
according to the workflow ' s output_type configuration. The length
of the returned list equals the length of the input tasks list .
Processing Flow :
1. Tasks are processed sequentially ( not concurrently with each other )
2. For each task , all agents execute concurrently
3. Results are collected and formatted for each task
4. Conversation history accumulates across all tasks
5. Final result list contains aggregated outputs for each task
Image Handling :
- If imgs is None : All tasks execute without images
- If imgs has fewer items than tasks : Extra tasks execute without images
- If imgs has more items than tasks : Extra images are ignored
- Each task gets at most one corresponding image
Dashboard Behavior :
- If show_dashboard = True , dashboard resets and updates for each task
- Progress is shown separately for each task ' s agent execution
- Final dashboard shows results from the last task only
Memory and Performance :
- Conversation history grows with each task ( cumulative )
- Memory usage scales with number of tasks and agent outputs
- CPU utilization is optimal during each task ' s concurrent execution
- Consider clearing conversation history for very large batch jobs
Example :
Financial analysis across quarters :
>> > workflow = ConcurrentWorkflow ( agents = [ analyst1 , analyst2 , analyst3 ] )
>> > quarterly_tasks = [
. . . " Analyze Q1 financial performance and market position " ,
. . . " Analyze Q2 financial performance and market position " ,
. . . " Analyze Q3 financial performance and market position " ,
. . . " Analyze Q4 financial performance and market position "
. . . ]
>> > results = workflow . batch_run ( quarterly_tasks )
>> > print ( f " Completed { len ( results ) } quarterly analyses " )
>> > # Each result contains insights from all 3 analysts for that quarter
With corresponding images :
>> > tasks = [ " Analyze chart trends " , " Review performance metrics " ]
>> > charts = [ " q1_chart.png " , " q2_chart.png " ]
>> > results = workflow . batch_run ( tasks , imgs = charts )
>> > # Task 0 uses q1_chart.png, Task 1 uses q2_chart.png
Batch processing with dashboard :
>> > workflow = ConcurrentWorkflow (
. . . agents = [ agent1 , agent2 ] ,
. . . show_dashboard = True
. . . )
>> > results = workflow . batch_run ( [
. . . " Process document batch 1 " ,
. . . " Process document batch 2 " ,
. . . " Process document batch 3 "
. . . ] )
# Dashboard shows progress for each batch separately
Use Cases :
- Multi - period financial analysis ( quarterly / yearly reports )
- Batch document processing with multiple expert reviews
- A / B testing across different scenarios or datasets
- Systematic evaluation of multiple proposals or options
- Comparative analysis across time periods or categories
Note :
Tasks are intentionally processed sequentially to maintain result
order and prevent resource contention . For truly concurrent task
processing , create separate workflow instances .
See Also :
- run ( ) : For single task execution
- ConcurrentWorkflow : For configuration options
"""
""" Execute workflow on multiple tasks sequentially. """
results = [ ]
for idx , task in enumerate ( tasks ) :
img = None
if imgs is not None :
# Use the img at the same index if available, else None
if idx < len ( imgs ) :
img = imgs [ idx ]
results . append ( self . run ( task = task , img = img ) )
if imgs is not None and idx < len ( imgs ) :
img = imgs [ idx ]
results . append (
self . run (
task = task ,
img = img ,
streaming_callback = streaming_callback ,
)
)
return results
# if __name__ == "__main__":
# # Assuming you've already initialized some agents outside of this class
# agents = [
# Agent(
# agent_name=f"Financial-Analysis-Agent-{i}",
# system_prompt=FINANCIAL_AGENT_SYS_PROMPT,
# model_name="gpt-4o",
# max_loops=1,
# )
# for i in range(3) # Adjust number of agents as needed
# ]
# # Initialize the workflow with the list of agents
# workflow = ConcurrentWorkflow(
# agents=agents,
# metadata_output_path="agent_metadata_4.json",
# return_str_on=True,
# )
# # Define the task for all agents
# task = "How can I establish a ROTH IRA to buy stocks and get a tax break? What are the criteria?"
# # Run the workflow and save metadata
# metadata = workflow.run(task)
# print(metadata)
Kye Gomez
2 months ago
