Concurrent Workflow

The ConcurrentWorkflow runs multiple agents simultaneously on the same task, enabling parallel execution and high-throughput processing. This architecture is ideal when you need multiple perspectives or rapid parallel analysis.

When to Use

High-throughput tasks: Process large volumes simultaneously
Multiple perspectives: Get diverse viewpoints on the same input
Parallel analysis: Market, financial, and risk analysis at once
Time-critical operations: Minimize total execution time
Independent processing: Tasks with no dependencies

Key Features

True parallel execution with ThreadPoolExecutor
Real-time dashboard monitoring (optional)
Agent status tracking
Streaming callbacks support
Thread pool sized by agent count, capped at 32
Conversation history aggregation, with a fresh conversation for every run
Results recorded in agent order, one final answer per agent

Basic Example

from swarms import Agent, ConcurrentWorkflow

# Create specialized analysts
market_analyst = Agent(
    agent_name="Market-Analyst",
    system_prompt="Analyze market trends and opportunities.",
    model_name="gpt-5.4",
)

financial_analyst = Agent(
    agent_name="Financial-Analyst",
    system_prompt="Provide financial analysis and projections.",
    model_name="gpt-5.4",
)

risk_analyst = Agent(
    agent_name="Risk-Analyst",
    system_prompt="Assess and quantify potential risks.",
    model_name="gpt-5.4",
)

# Create concurrent workflow
workflow = ConcurrentWorkflow(
    agents=[market_analyst, financial_analyst, risk_analyst],
    max_loops=1,
)

# All agents run simultaneously
results = workflow.run(
    "Analyze the potential impact of AI on the healthcare industry"
)
print(results)

With Dashboard Monitoring

workflow = ConcurrentWorkflow(
    name="AnalysisTeam",
    description="Concurrent analysis team",
    agents=[market_analyst, financial_analyst, risk_analyst],
    show_dashboard=True,  # Enable real-time dashboard
    max_loops=1,
)

results = workflow.run("Evaluate cryptocurrency market trends")

With Streaming Callbacks

def streaming_callback(agent_name: str, chunk: str, is_final: bool):
    """Receive real-time updates from agents"""
    if is_final:
        print(f"\n[{agent_name}] Complete!")
    else:
        print(f"[{agent_name}]: {chunk}", end="", flush=True)

results = workflow.run(
    task="Analyze Q4 performance",
    streaming_callback=streaming_callback,
)

Key Parameters

str

default:"ConcurrentWorkflow"

Name identifier for the workflow

List[Agent]

required

List of agents to execute concurrently

int

default:"1"

Reserved for future use — currently stored but not applied. Each call to run() executes every agent exactly once regardless of this value; use batch_run() to process multiple tasks.

bool

default:"False"

Enable real-time dashboard display

str

default:"dict-all-except-first"

Output format for results. The default returns each agent’s message as a {"role", "content"} dict, without the user’s task.

bool

default:"False"

Accepted and stored, but never read. Setting it has no effect.

bool

default:"True"

Automatically save conversation history to the workflow workspace directory after each run

ConcurrentWorkflow.__init__ accepts two similarly named parameters: autosave and auto_save (both default True). Only autosave has any effect — it is the flag that enables the workspace writer. auto_save is stored on the instance and never read, so setting auto_save=False does not turn saving off. Always use autosave.

bool

default:"False"

Emit debug-level logging for workspace and autosave operations

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.

Methods

run()

Execute all agents concurrently on a task. Each call starts with a fresh conversation, so a reused workflow returns only this task’s messages.

result = workflow.run(
    task="Analyze market conditions",
    img=None,  # Optional image input
    streaming_callback=None,  # Optional callback
)

batch_run()

Process multiple tasks sequentially (each task runs agents concurrently). Every task runs in its own conversation, so a result never includes messages from earlier tasks.

tasks = [
    "Analyze tech sector",
    "Analyze healthcare sector",
    "Analyze energy sector"
]

results = workflow.batch_run(tasks)

Dashboard Features

When show_dashboard=True, you get:

Real-time Status: See each agent’s current state (pending, running, completed)
Output Preview: Monitor agent outputs as they generate
Progress Tracking: Visual progress indicators
Error Detection: Immediate error visibility
Completion Summary: Final dashboard with all results

Use Cases

Multi-Perspective Analysis

# Get technical, fundamental, and sentiment analysis simultaneously
technical_agent = Agent(agent_name="Technical-Analyst", ...)
fundamental_agent = Agent(agent_name="Fundamental-Analyst", ...)
sentiment_agent = Agent(agent_name="Sentiment-Analyst", ...)

analysis_team = ConcurrentWorkflow(
    agents=[technical_agent, fundamental_agent, sentiment_agent]
)

insights = analysis_team.run("NVIDIA stock analysis")

Parallel Research

# Research multiple aspects simultaneously
literature_researcher = Agent(agent_name="Literature-Researcher", ...)
market_researcher = Agent(agent_name="Market-Researcher", ...)
competitor_researcher = Agent(agent_name="Competitor-Researcher", ...)

research_team = ConcurrentWorkflow(
    agents=[literature_researcher, market_researcher, competitor_researcher]
)

research = research_team.run("Electric vehicle market landscape")

Batch Document Processing

# Process documents with multiple analyzers
summary_agent = Agent(agent_name="Summarizer", ...)
sentiment_agent = Agent(agent_name="Sentiment-Analyzer", ...)
key_points_agent = Agent(agent_name="Key-Points-Extractor", ...)

processing_team = ConcurrentWorkflow(
    agents=[summary_agent, sentiment_agent, key_points_agent]
)

# Tasks are plain strings, so pass each document's text rather than its path
documents = [open(path).read() for path in ["doc1.txt", "doc2.txt", "doc3.txt"]]
results = processing_team.batch_run(documents)

Performance Optimization

Thread pool sizing

The thread pool is sized by agent count, capped at 32 (MAX_CONCURRENT_AGENTS), since agent calls are network-bound rather than CPU-bound. Pass max_workers to override it, for example to stay under a provider’s rate limit:

workflow = ConcurrentWorkflow(
    agents=agents,
    max_workers=4,
)

Agent Configuration for Concurrency

# Disable printing for dashboard mode
if show_dashboard:
    for agent in agents:
        agent.print_on = False

Advanced Features

Agent Status Tracking

workflow = ConcurrentWorkflow(
    agents=[agent1, agent2, agent3],
    show_dashboard=True,
)

# Internal status tracking for each agent
# - status: "pending" | "running" | "completed" | "error"
# - output: Current agent output
print(workflow.agent_statuses)

Conversation Aggregation

result = workflow.run("Task")

# Access the latest run's conversation
history = workflow.conversation.conversation_history

# Format:
# [{"role": "User", "content": "Task"},
#  {"role": "Agent1", "content": "Response1"},
#  {"role": "Agent2", "content": "Response2"}]

Each agent contributes its final answer, not its whole transcript, and messages follow the order of agents rather than completion order.

Output Types

Supported output formats include:

"dict-all-except-first" (default): List of message dicts, without the user’s task
"dict" / "list": List of every message dict, task included
"str" / "all": The transcript as "role: content" text
"str-all-except-first": The agents’ answers as text, without the task
"final": The last agent’s answer

workflow = ConcurrentWorkflow(
    agents=agents,
    output_type="dict",
)

Best Practices

Performance Tip: Use concurrent workflow when agents can truly work independently

Independent Agents: Ensure agents don’t need each other’s outputs
Appropriate Size: 3-8 agents typically optimal for most systems
Dashboard Usage: Enable for debugging, disable for production
Resource Management: Monitor CPU/memory with large agent counts
Error Handling: With the default on_error="store", one agent failure doesn’t stop others

Concurrent execution means agents run simultaneously - ensure your LLM API can handle parallel requests

Error Handling

try:
    results = workflow.run("Analyze market")
except Exception as e:
    print(f"Workflow error: {e}")
    # With on_error="store", agent errors appear in results as "<name> (failed)"

Sequential Workflow - For ordered execution
Mixture of Agents - For synthesis of parallel outputs
Agent Rearrange - For mixed patterns

Getting Started

Core Concepts

Agent Development

Reasoning Agents

Multi-Agent Architectures

Deployment & Operations

Integrations

CLI Reference

Changelog

Concurrent Workflow

When to Use

Key Features

Basic Example

With Dashboard Monitoring

With Streaming Callbacks

Key Parameters

Methods

run()

batch_run()

Dashboard Features

Use Cases

Multi-Perspective Analysis

Parallel Research

Batch Document Processing

Performance Optimization

Thread pool sizing

Agent Configuration for Concurrency

Advanced Features

Agent Status Tracking

Conversation Aggregation

Output Types

Best Practices

Error Handling

​When to Use

​Key Features

​Basic Example

​With Dashboard Monitoring

​With Streaming Callbacks

​Key Parameters

​Methods

​run()

​batch_run()

​Dashboard Features

​Use Cases

​Multi-Perspective Analysis

​Parallel Research

​Batch Document Processing

​Performance Optimization

​Thread pool sizing

​Agent Configuration for Concurrency

​Advanced Features

​Agent Status Tracking

​Conversation Aggregation

​Output Types

​Best Practices

​Error Handling

​Related Architectures

When to Use

Key Features

Basic Example

With Dashboard Monitoring

With Streaming Callbacks

Key Parameters

Methods

run()

batch_run()

Dashboard Features

Use Cases

Multi-Perspective Analysis

Parallel Research

Batch Document Processing

Performance Optimization

Thread pool sizing

Agent Configuration for Concurrency

Advanced Features

Agent Status Tracking

Conversation Aggregation

Output Types

Best Practices

Error Handling

Related Architectures