ConcurrentWorkflow

Overview

The ConcurrentWorkflow class provides a framework for executing multiple agents concurrently on the same task, with optional dashboard monitoring, streaming callbacks, and various output formatting options. It uses ThreadPoolExecutor to manage concurrent execution and provides real-time status tracking for each agent.

Key Features

Concurrent Execution: Run multiple agents simultaneously on the same task
Real-time Dashboard: Monitor agent status and outputs in real-time
Streaming Callbacks: Get real-time updates as agents generate outputs
Flexible Output Formatting: Multiple output format options
Auto-save Support: Automatically save conversation history
Error Handling: Graceful error handling with status tracking
Batch Processing: Process multiple tasks sequentially with concurrent agents

Installation

pip install -U swarms

Class Definition

class ConcurrentWorkflow:
    def __init__(
        self,
        id: str = None,
        name: str = "ConcurrentWorkflow",
        description: str = "Execution of multiple agents concurrently",
        agents: List[Union[Agent, Callable]] = None,
        auto_save: bool = True,
        output_type: str = "dict-all-except-first",
        max_loops: int = 1,
        auto_generate_prompts: bool = False,
        show_dashboard: bool = False,
        autosave: bool = True,
        verbose: bool = False,
        on_error: str = "store",
        max_workers: Optional[int] = None,
    )

Parameters

str

default:"auto-generated"

Unique identifier for the workflow instance. Auto-generated if not provided.

str

default:"ConcurrentWorkflow"

Human-readable name for the workflow

str

default:"Execution of multiple agents concurrently"

Description of the workflow’s purpose

List[Union[Agent, Callable]]

required

List of agents to execute concurrently. Must not be None or empty.

bool

default:"True"

Accepted and stored, but never read. Setting it has no effect; use autosave to control saving.

str

default:"dict-all-except-first"

Format of the returned conversation. Common values: "dict-all-except-first" (every agent’s message as a {"role", "content"} dict, without the user’s task), "dict" or "list" (every message, task included), "str" or "all" (the transcript as "role: content" text), "str-all-except-first" (the agents’ answers as text) and "final" (the last agent’s answer). See SequentialWorkflow for the full list.

int

default:"1"

Maximum number of execution loops (currently unused in concurrent execution)

bool

default:"False"

Accepted and stored, but never read — setting it has no effect. The activate_auto_prompt_engineering() method that once consumed it no longer exists.

bool

default:"False"

Whether to display real-time dashboard during execution

bool

default:"True"

Whether to automatically save conversation history to workspace

str

default:"store"

How to handle an agent that raises. "store" records the error as that agent’s output and lets the rest finish; "raise" propagates the first error and aborts the run.

int

default:"None"

Thread pool size. Defaults to len(agents) capped at 32. Agent calls are network-bound, so this is sized by agent count rather than CPU cores.

bool

default:"False"

Whether to enable verbose logging

Methods

`run(task, img=None, imgs=None, streaming_callback=None)`

Execute all agents concurrently on the given task. Each call starts with a fresh conversation, so a reused workflow returns only this task’s messages. Each agent’s final answer is recorded, not its whole transcript, and answers are collected in the order of agents, not completion order. With on_error="store", a failed agent is recorded with "Error: ..." as its content, under the role "<agent_name> (failed)" (or its plain name in dashboard mode). The conversation is autosaved after the run, even when it fails.

str

required

The task to be executed by all agents

Optional[str]

default:"None"

Single image path for agents that support image input

Optional[List[str]]

default:"None"

List of image paths for agents that support multiple images

Optional[Callable[[str, str, bool], None]]

default:"None"

Callback function for streaming updates. Called with (agent_name, chunk, is_final) parameters.

Union[Dict, List, str]

Formatted conversation history based on output_type

`batch_run(tasks, imgs=None, streaming_callback=None)`

Execute workflow on multiple tasks sequentially, running every agent concurrently on each task. Every task runs in its own conversation, so each result holds only that task’s messages. The workflow’s own conversation is restored afterwards and is left untouched by the batch.

List[str]

required

List of tasks to be executed

Optional[List[str]]

default:"None"

List of image paths, matched to tasks by index. Tasks past the end of the list run without an image.

Optional[Callable[[str, str, bool], None]]

default:"None"

Callback function for streaming updates

List[Union[Dict, List, str]]

List of results for each task

`run_with_dashboard(task, img=None, imgs=None, streaming_callback=None)`

Execute agents with real-time dashboard monitoring.

str

required

The task to be executed by all agents

Optional[str]

default:"None"

Single image path for agents that support image input

Optional[List[str]]

default:"None"

List of image paths for agents that support multiple images

Optional[Callable[[str, str, bool], None]]

default:"None"

Callback function for streaming updates

Union[Dict, List, str]

Formatted conversation history based on output_type

`fix_agents()`

Configure agents for dashboard mode. Disables printing (agent.print_on = False) for every agent when show_dashboard=True, to prevent console output from conflicting with the dashboard display. Called automatically during initialization when show_dashboard=True.

List[Union[Agent, Callable]]

The configured list of agents

`reliability_check()`

Validate workflow configuration. Raises:

ValueError: If no agents are provided or agents list is empty

`display_agent_dashboard(title="ConcurrentWorkflow Dashboard", is_final=False)`

Display real-time dashboard showing agent status and outputs.

str

default:"ConcurrentWorkflow Dashboard"

Title to display for the dashboard

bool

default:"False"

Whether this is the final dashboard display

Attributes

Attribute	Type	Description
`id`	str	Unique identifier for the workflow instance
`name`	str	Human-readable name for the workflow
`description`	str	Description of the workflow’s purpose
`agents`	List[Union[Agent, Callable]]	List of agents to execute concurrently
`agent_statuses`	dict	Status (`"pending"`, `"running"`, `"completed"`, `"error"`) and output of each agent, updated in dashboard mode
`conversation`	Conversation	The latest run’s conversation; replaced at the start of every `run()`
`on_error`	str	Failure policy, `"store"` or `"raise"`
`max_workers`	Optional[int]	Thread pool size override
`workspace`	WorkspaceManager	Owns the autosave directory
`swarm_workspace_dir`	Optional[str]	Autosave directory; `None` when `autosave=False`
`metadata_output_path`	str	Metadata file name built from the workflow name and ID; nothing is written to it

Usage Examples

Basic Concurrent Workflow

from swarms import Agent, ConcurrentWorkflow

# Create specialized agents with different perspectives
technical_analyst = Agent(
    agent_name="Technical-Analyst",
    model_name="claude-sonnet-4-6",
    temperature=0.5,
    max_loops=1,
    system_prompt="You are a technical analyst. Analyze from a technical perspective."
)

business_analyst = Agent(
    agent_name="Business-Analyst",
    model_name="claude-sonnet-4-6",
    temperature=0.5,
    max_loops=1,
    system_prompt="You are a business analyst. Analyze from a business perspective."
)

user_researcher = Agent(
    agent_name="User-Researcher",
    model_name="claude-sonnet-4-6",
    temperature=0.5,
    max_loops=1,
    system_prompt="You are a user researcher. Analyze from a user experience perspective."
)

# Create concurrent workflow
workflow = ConcurrentWorkflow(
    name="Multi-Perspective Analysis",
    description="Analyze from multiple perspectives simultaneously",
    agents=[technical_analyst, business_analyst, user_researcher],
    verbose=True
)

# Run all agents concurrently on the same task
result = workflow.run("Analyze the potential of implementing AI chatbots in customer service")
print(result)

With Real-Time Dashboard

# Create workflow with dashboard enabled
workflow = ConcurrentWorkflow(
    name="Analysis with Dashboard",
    agents=[technical_analyst, business_analyst, user_researcher],
    show_dashboard=True,  # Enable real-time dashboard
    verbose=True
)

# Run and watch the dashboard update in real-time
result = workflow.run("Evaluate the impact of remote work on productivity")

With Streaming Callbacks

def handle_stream(agent_name: str, chunk: str, is_final: bool):
    """Custom streaming callback to handle real-time updates"""
    if is_final:
        print(f"\n[{agent_name}] Completed!")
    else:
        print(f"[{agent_name}] {chunk}", end="", flush=True)

workflow = ConcurrentWorkflow(
    name="Streaming Analysis",
    agents=[technical_analyst, business_analyst],
)

# Get real-time streaming updates
result = workflow.run(
    "Analyze cybersecurity trends",
    streaming_callback=handle_stream
)

Batch Processing

# Process multiple tasks sequentially, with agents running concurrently on each
tasks = [
    "Analyze the impact of AI on healthcare",
    "Evaluate renewable energy solutions",
    "Assess blockchain technology trends"
]

workflow = ConcurrentWorkflow(
    name="Batch Analysis",
    agents=[technical_analyst, business_analyst, user_researcher],
)

results = workflow.batch_run(tasks)

for i, (task, result) in enumerate(zip(tasks, results), 1):
    print(f"\nTask {i}: {task}")
    print(f"Result: {result}")
    print("-" * 80)

With Image Input

from swarms import Agent, ConcurrentWorkflow

# claude-sonnet-4-6 is vision-capable — no separate vision wrapper needed
image_analyst1 = Agent(
    agent_name="Image-Analyst-1",
    model_name="claude-sonnet-4-6",
    system_prompt="Analyze images for technical details"
)

image_analyst2 = Agent(
    agent_name="Image-Analyst-2",
    model_name="claude-sonnet-4-6",
    system_prompt="Analyze images for aesthetic quality"
)

workflow = ConcurrentWorkflow(
    agents=[image_analyst1, image_analyst2],
)

# Analyze image with multiple agents concurrently
result = workflow.run(
    "Analyze this product image",
    img="path/to/product.jpg"
)

Custom Output Types

# Different output formats
workflow_dict = ConcurrentWorkflow(
    agents=[technical_analyst, business_analyst],
    output_type="dict"  # Every message, including the user's task
)

workflow_str = ConcurrentWorkflow(
    agents=[technical_analyst, business_analyst],
    output_type="str"  # Transcript as "role: content" text
)

workflow_dict_except_first = ConcurrentWorkflow(
    agents=[technical_analyst, business_analyst],
    output_type="dict-all-except-first"  # Only the agents' messages, task dropped
)

With Autosave

import os

# Set workspace directory
os.environ["WORKSPACE_DIR"] = "./analysis_workspace"

workflow = ConcurrentWorkflow(
    name="Saved-Analysis",
    agents=[technical_analyst, business_analyst],
    autosave=True,  # Save conversation history
    verbose=True
)

result = workflow.run("Analyze market trends")
# Saved to ./analysis_workspace/swarms/ConcurrentWorkflow/Saved-Analysis-<timestamp>/conversation_history.json

Error Handling

try:
    workflow = ConcurrentWorkflow(
        agents=[technical_analyst, business_analyst],
        show_dashboard=True
    )
    result = workflow.run("Analyze this topic")
except ValueError as e:
    print(f"Configuration error: {e}")
except Exception as e:
    print(f"Execution error: {e}")

With the default on_error="store", a failing agent does not raise: its error is recorded in the result and the other agents finish. Set on_error="raise" to abort the run on the first agent error instead.

Dashboard Output Example

When show_dashboard=True, you’ll see real-time updates like:

╔═══════════════════════════════════════════════════════════════╗
║           ConcurrentWorkflow Dashboard                         ║
╠═══════════════════════════════════════════════════════════════╣
║ Agent: Technical-Analyst                                       ║
║ Status: running                                                ║
║ Output: Analyzing technical aspects...                         ║
╠═══════════════════════════════════════════════════════════════╣
║ Agent: Business-Analyst                                        ║
║ Status: completed                                              ║
║ Output: The business impact is significant...                  ║
╠═══════════════════════════════════════════════════════════════╣
║ Agent: User-Researcher                                         ║
║ Status: pending                                                ║
║ Output:                                                        ║
╚═══════════════════════════════════════════════════════════════╝

Best Practices

Agent Diversity: Use agents with different perspectives for richer analysis
Dashboard for Monitoring: Enable dashboard during development to monitor agent progress
Streaming Callbacks: Use streaming callbacks for real-time feedback in production
Error Handling: Always wrap concurrent execution in try-except blocks
Resource Management: Be mindful of API rate limits when running many agents
Task Design: Ensure the task benefits from multiple concurrent perspectives
Autosave: Enable autosave for important analyses

Common Use Cases

Multi-Perspective Analysis: Get technical, business, and user perspectives simultaneously
Consensus Building: Run multiple agents and aggregate their outputs
Parallel Research: Research different aspects of a topic concurrently
Voting Systems: Multiple agents vote on decisions
Quality Assurance: Multiple agents review the same content
Competitive Analysis: Different agents analyze competing solutions

Performance Considerations

Concurrency: Thread pool sized by agent count, capped at 32 (MAX_CONCURRENT_AGENTS). Agent calls are network-bound, so this does not derive from CPU core count. Override with max_workers.
Concurrent Execution: All agents run truly concurrently using ThreadPoolExecutor
Memory Usage: Each agent maintains its own context, consider memory for many agents
API Rate Limits: Be aware of rate limits when using cloud-based LLMs

SequentialWorkflow: For sequential agent execution
AgentRearrange: For complex orchestration patterns
GraphWorkflow: For DAG-based workflows
Agent: The base agent class used in workflows

Core Classes

Multi-Agent Harnesses

Tools & Utilities

Deployment

ConcurrentWorkflow

Overview

Key Features

Installation

Class Definition

Parameters

Methods

`run(task, img=None, imgs=None, streaming_callback=None)`

`batch_run(tasks, imgs=None, streaming_callback=None)`

`run_with_dashboard(task, img=None, imgs=None, streaming_callback=None)`

`fix_agents()`

`reliability_check()`

`display_agent_dashboard(title="ConcurrentWorkflow Dashboard", is_final=False)`

Attributes

Usage Examples

Basic Concurrent Workflow

With Real-Time Dashboard

With Streaming Callbacks

Batch Processing

With Image Input

Custom Output Types

With Autosave

Error Handling

Dashboard Output Example

Best Practices

Common Use Cases

Performance Considerations

​Overview

​Key Features

​Installation

​Class Definition

​Parameters

​Methods

​run(task, img=None, imgs=None, streaming_callback=None)

​batch_run(tasks, imgs=None, streaming_callback=None)

​run_with_dashboard(task, img=None, imgs=None, streaming_callback=None)

​fix_agents()

​reliability_check()

​display_agent_dashboard(title="ConcurrentWorkflow Dashboard", is_final=False)

​Attributes

​Usage Examples

​Basic Concurrent Workflow

​With Real-Time Dashboard

​With Streaming Callbacks

​Batch Processing

​With Image Input

​Custom Output Types

​With Autosave

​Error Handling

​Dashboard Output Example

​Best Practices

​Common Use Cases

​Performance Considerations

​Related Classes

Overview

Key Features

Installation

Class Definition

Parameters

Methods

`run(task, img=None, imgs=None, streaming_callback=None)`

`batch_run(tasks, imgs=None, streaming_callback=None)`

`run_with_dashboard(task, img=None, imgs=None, streaming_callback=None)`

`fix_agents()`

`reliability_check()`

`display_agent_dashboard(title="ConcurrentWorkflow Dashboard", is_final=False)`

Attributes

Usage Examples

Basic Concurrent Workflow

With Real-Time Dashboard

With Streaming Callbacks

Batch Processing

With Image Input

Custom Output Types

With Autosave

Error Handling

Dashboard Output Example

Best Practices

Common Use Cases

Performance Considerations

Related Classes