Multi-Agent Execution API Reference¶

This comprehensive documentation covers all functions in the multi_agent_exec.py module for running multiple agents using various execution strategies. The module provides synchronous and asynchronous execution methods, optimized performance with uvloop, and utility functions for information retrieval.

Function Overview¶

Function	Signature	Category	Description
`run_single_agent`	`run_single_agent(agent, task, args, *kwargs) -> Any`	Single Agent	Runs a single agent synchronously
`run_agent_async`	`run_agent_async(agent, task) -> Any`	Single Agent	Runs a single agent asynchronously using asyncio
`run_agents_concurrently_async`	`run_agents_concurrently_async(agents, task) -> List[Any]`	Concurrent Execution	Runs multiple agents concurrently using asyncio
`run_agents_concurrently`	`run_agents_concurrently(agents, task, max_workers=None) -> List[Any]`	Concurrent Execution	Optimized concurrent agent runner using ThreadPoolExecutor
`run_agents_concurrently_multiprocess`	`run_agents_concurrently_multiprocess(agents, task, batch_size=None) -> List[Any]`	Concurrent Execution	Manages agents concurrently in batches with optimized performance
`batched_grid_agent_execution`	`batched_grid_agent_execution(agents, tasks, max_workers=None) -> List[Any]`	Batched & Grid	Runs multiple agents with different tasks concurrently
`run_agents_with_different_tasks`	`run_agents_with_different_tasks(agent_task_pairs, batch_size=10, max_workers=None) -> List[Any]`	Batched & Grid	Runs agents with different tasks concurrently in batches
`run_agents_concurrently_uvloop`	`run_agents_concurrently_uvloop(agents, task, max_workers=None) -> List[Any]`	UVLoop Optimized	Runs agents concurrently using uvloop for optimized performance
`run_agents_with_tasks_uvloop`	`run_agents_with_tasks_uvloop(agents, tasks, max_workers=None) -> List[Any]`	UVLoop Optimized	Runs agents with different tasks using uvloop optimization
`get_swarms_info`	`get_swarms_info(swarms) -> str`	Utility	Fetches and formats information about available swarms
`get_agents_info`	`get_agents_info(agents, team_name=None) -> str`	Utility	Fetches and formats information about available agents

Single Agent Functions¶

`run_single_agent(agent, task, *args, **kwargs)`¶

Runs a single agent synchronously.

Signature¶

def run_single_agent(
    agent: AgentType,
    task: str,
    *args,
    **kwargs
) -> Any

Parameters¶

Parameter	Type	Required	Description
`agent`	`AgentType`	Yes	Agent instance to run
`task`	`str`	Yes	Task string to execute
`*args`	`Any`	No	Additional positional arguments
`**kwargs`	`Any`	No	Additional keyword arguments

Returns¶

Any: Agent execution result

Example¶

from swarms.structs.agent import Agent
from swarms.structs.multi_agent_exec import run_single_agent

agent = Agent(
    agent_name="Financial-Analyst",
    system_prompt="You are a financial analysis expert",
    model_name="gpt-4o-mini",
    max_loops=1
)

result = run_single_agent(agent, "Analyze the current stock market trends")
print(result)

`run_agent_async(agent, task)`¶

Runs a single agent asynchronously using asyncio.

Signature¶

async def run_agent_async(agent: AgentType, task: str) -> Any

Parameters¶

Parameter	Type	Required	Description
`agent`	`AgentType`	Yes	Agent instance to run
`task`	`str`	Yes	Task string to execute

Returns¶

Any: Agent execution result

Example¶

import asyncio
from swarms.structs.agent import Agent
from swarms.structs.multi_agent_exec import run_agent_async

async def main():
    agent = Agent(
        agent_name="Researcher",
        system_prompt="You are a research assistant",
        model_name="gpt-4o-mini",
        max_loops=1
    )

    result = await run_agent_async(agent, "Research AI advancements in 2024")
    print(result)

asyncio.run(main())

Concurrent Execution Functions¶

`run_agents_concurrently_async(agents, task)`¶

Runs multiple agents concurrently using asyncio.

Signature¶

async def run_agents_concurrently_async(
    agents: List[AgentType],
    task: str
) -> List[Any]

Parameters¶

Parameter	Type	Required	Description
`agents`	`List[AgentType]`	Yes	List of Agent instances to run concurrently
`task`	`str`	Yes	Task string to execute by all agents

Returns¶

List[Any]: List of outputs from each agent

Example¶

import asyncio
from swarms.structs.agent import Agent
from swarms.structs.multi_agent_exec import run_agents_concurrently_async

async def main():
    agents = [
        Agent(
            agent_name=f"Analyst-{i}",
            system_prompt="You are a market analyst",
            model_name="gpt-4o-mini",
            max_loops=1
        )
        for i in range(3)
    ]

    task = "Analyze the impact of AI on job markets"
    results = await run_agents_concurrently_async(agents, task)

    for i, result in enumerate(results):
        print(f"Agent {i+1} result: {result}")

asyncio.run(main())

`run_agents_concurrently(agents, task, max_workers=None)`¶

Optimized concurrent agent runner using ThreadPoolExecutor.

Signature¶

def run_agents_concurrently(
    agents: List[AgentType],
    task: str,
    max_workers: Optional[int] = None,
) -> List[Any]

Parameters¶

Parameter	Type	Required	Default	Description
`agents`	`List[AgentType]`	Yes	-	List of Agent instances to run concurrently
`task`	`str`	Yes	-	Task string to execute
`max_workers`	`Optional[int]`	No	95% of CPU cores	Maximum number of threads in the executor

Returns¶

List[Any]: List of outputs from each agent (exceptions included if agents fail)

Example¶

from swarms.structs.agent import Agent
from swarms.structs.multi_agent_exec import run_agents_concurrently

# Create multiple agents
agents = [
    Agent(
        agent_name="Tech-Analyst",
        system_prompt="You are a technology analyst",
        model_name="gpt-4o-mini",
        max_loops=1
    ),
    Agent(
        agent_name="Finance-Analyst",
        system_prompt="You are a financial analyst",
        model_name="gpt-4o-mini",
        max_loops=1
    ),
    Agent(
        agent_name="Market-Strategist",
        system_prompt="You are a market strategist",
        model_name="gpt-4o-mini",
        max_loops=1
    )
]

task = "Analyze the future of electric vehicles in 2025"
results = run_agents_concurrently(agents, task, max_workers=4)

for i, result in enumerate(results):
    print(f"Agent {i+1} ({agents[i].agent_name}): {result}")

`run_agents_concurrently_multiprocess(agents, task, batch_size=None)`¶

Manages and runs multiple agents concurrently in batches with optimized performance.

Signature¶

def run_agents_concurrently_multiprocess(
    agents: List[Agent],
    task: str,
    batch_size: int = os.cpu_count()
) -> List[Any]

Parameters¶

Parameter	Type	Required	Default	Description
`agents`	`List[Agent]`	Yes	-	List of Agent instances to run concurrently
`task`	`str`	Yes	-	Task string to execute by all agents
`batch_size`	`int`	No	CPU count	Number of agents to run in parallel in each batch

Returns¶

List[Any]: List of outputs from each agent

Example¶

import os
from swarms.structs.agent import Agent
from swarms.structs.multi_agent_exec import run_agents_concurrently_multiprocess

agents = [
    Agent(
        agent_name=f"Research-Agent-{i}",
        system_prompt="You are a research specialist",
        model_name="gpt-4o-mini",
        max_loops=1
    )
    for i in range(5)
]

task = "Research the benefits of renewable energy"
batch_size = os.cpu_count()  # Use all CPU cores
results = run_agents_concurrently_multiprocess(agents, task, batch_size)

print(f"Completed {len(results)} agent executions")

Batched and Grid Execution¶

`batched_grid_agent_execution(agents, tasks, max_workers=None)`¶

Runs multiple agents with different tasks concurrently using batched grid execution.

Signature¶

def batched_grid_agent_execution(
    agents: List["AgentType"],
    tasks: List[str],
    max_workers: int = None,
) -> List[Any]

Parameters¶

Parameter	Type	Required	Default	Description
`agents`	`List[AgentType]`	Yes	-	List of agent instances
`tasks`	`List[str]`	Yes	-	List of tasks, one for each agent
`max_workers`	`int`	No	90% of CPU cores	Maximum number of threads to use

Returns¶

List[Any]: List of results from each agent

Raises¶

ValueError: If number of agents doesn't match number of tasks

Example¶

from swarms.structs.agent import Agent
from swarms.structs.multi_agent_exec import batched_grid_agent_execution

agents = [
    Agent(
        agent_name="Data-Scientist",
        system_prompt="You are a data science expert",
        model_name="gpt-4o-mini",
        max_loops=1
    ),
    Agent(
        agent_name="ML-Engineer",
        system_prompt="You are a machine learning engineer",
        model_name="gpt-4o-mini",
        max_loops=1
    ),
    Agent(
        agent_name="AI-Researcher",
        system_prompt="You are an AI researcher",
        model_name="gpt-4o-mini",
        max_loops=1
    )
]

tasks = [
    "Analyze machine learning algorithms performance",
    "Design a neural network architecture",
    "Research latest AI breakthroughs"
]

results = batched_grid_agent_execution(agents, tasks, max_workers=3)

for i, result in enumerate(results):
    print(f"Task {i+1}: {tasks[i]}")
    print(f"Result: {result}\n")

`run_agents_with_different_tasks(agent_task_pairs, batch_size=10, max_workers=None)`¶

Runs multiple agents with different tasks concurrently, processing them in batches.

Signature¶

def run_agents_with_different_tasks(
    agent_task_pairs: List[tuple["AgentType", str]],
    batch_size: int = 10,
    max_workers: int = None,
) -> List[Any]

Parameters¶

Parameter	Type	Required	Default	Description
`agent_task_pairs`	`List[tuple[AgentType, str]]`	Yes	-	List of (agent, task) tuples
`batch_size`	`int`	No	10	Number of agents to run in parallel in each batch
`max_workers`	`int`	No	None	Maximum number of threads

Returns¶

List[Any]: List of outputs from each agent, in the same order as input pairs

Example¶

from swarms.structs.agent import Agent
from swarms.structs.multi_agent_exec import run_agents_with_different_tasks

# Create agents
agents = [
    Agent(
        agent_name="Content-Writer",
        system_prompt="You are a content writer",
        model_name="gpt-4o-mini",
        max_loops=1
    ),
    Agent(
        agent_name="Editor",
        system_prompt="You are an editor",
        model_name="gpt-4o-mini",
        max_loops=1
    ),
    Agent(
        agent_name="SEO-Specialist",
        system_prompt="You are an SEO specialist",
        model_name="gpt-4o-mini",
        max_loops=1
    )
]

# Create agent-task pairs
agent_task_pairs = [
    (agents[0], "Write a blog post about sustainable living"),
    (agents[1], "Edit and improve this article draft"),
    (agents[2], "Optimize this content for SEO")
]

results = run_agents_with_different_tasks(agent_task_pairs, batch_size=2)

for i, result in enumerate(results):
    agent, task = agent_task_pairs[i]
    print(f"{agent.agent_name} - {task}: {result}")

UVLoop Optimized Functions¶

`run_agents_concurrently_uvloop(agents, task, max_workers=None)`¶

Runs multiple agents concurrently using uvloop for optimized async performance.

Signature¶

def run_agents_concurrently_uvloop(
    agents: List[AgentType],
    task: str,
    max_workers: Optional[int] = None,
) -> List[Any]

Parameters¶

Parameter	Type	Required	Default	Description
`agents`	`List[AgentType]`	Yes	-	List of Agent instances to run concurrently
`task`	`str`	Yes	-	Task string to execute by all agents
`max_workers`	`Optional[int]`	No	95% of CPU cores	Maximum number of threads in the executor

Returns¶

List[Any]: List of outputs from each agent

Raises¶

ImportError: If uvloop is not installed
RuntimeError: If uvloop cannot be set as the event loop policy

Example¶

from swarms.structs.agent import Agent
from swarms.structs.multi_agent_exec import run_agents_concurrently_uvloop

# Note: uvloop must be installed (pip install uvloop)
agents = [
    Agent(
        agent_name="Performance-Analyst",
        system_prompt="You are a performance analyst",
        model_name="gpt-4o-mini",
        max_loops=1
    )
    for _ in range(5)
]

task = "Analyze system performance metrics"
results = run_agents_concurrently_uvloop(agents, task)

print(f"Processed {len(results)} agents with uvloop optimization")

`run_agents_with_tasks_uvloop(agents, tasks, max_workers=None)`¶

Runs multiple agents with different tasks concurrently using uvloop.

Signature¶

def run_agents_with_tasks_uvloop(
    agents: List[AgentType],
    tasks: List[str],
    max_workers: Optional[int] = None,
) -> List[Any]

Parameters¶

Parameter	Type	Required	Default	Description
`agents`	`List[AgentType]`	Yes	-	List of Agent instances to run
`tasks`	`List[str]`	Yes	-	List of task strings (must match number of agents)
`max_workers`	`Optional[int]`	No	95% of CPU cores	Maximum number of threads

Returns¶

List[Any]: List of outputs from each agent

Raises¶

ValueError: If number of agents doesn't match number of tasks
ImportError: If uvloop is not installed
RuntimeError: If uvloop cannot be set as the event loop policy

Example¶

from swarms.structs.agent import Agent
from swarms.structs.multi_agent_exec import run_agents_with_tasks_uvloop

agents = [
    Agent(
        agent_name="Data-Analyst-1",
        system_prompt="You are a data analyst",
        model_name="gpt-4o-mini",
        max_loops=1
    ),
    Agent(
        agent_name="Data-Analyst-2",
        system_prompt="You are a data analyst",
        model_name="gpt-4o-mini",
        max_loops=1
    )
]

tasks = [
    "Analyze sales data from Q1 2024",
    "Analyze customer satisfaction metrics"
]

results = run_agents_with_tasks_uvloop(agents, tasks)

for i, result in enumerate(results):
    print(f"Task: {tasks[i]}")
    print(f"Result: {result}\n")

Utility Functions¶

`get_swarms_info(swarms)`¶

Fetches and formats information about all available swarms in the system.

Signature¶

def get_swarms_info(swarms: List[Callable]) -> str

Parameters¶

Parameter	Type	Required	Description
`swarms`	`List[Callable]`	Yes	List of swarm objects to get information about

Returns¶

str: Formatted string containing names and descriptions of all swarms

Example¶

from swarms.structs.multi_agent_exec import get_swarms_info

# Assuming you have swarm objects
swarms = [
    # Your swarm objects here
]

info = get_swarms_info(swarms)
print(info)
# Output:
# Available Swarms:
#
# [Swarm 1]
# Name: ResearchSwarm
# Description: A swarm for research tasks
# Length of Agents: 3
# Swarm Type: hierarchical

`get_agents_info(agents, team_name=None)`¶

Fetches and formats information about all available agents in the system.

Signature¶

def get_agents_info(
    agents: List[Union[Agent, Callable]],
    team_name: str = None
) -> str

Parameters¶

Parameter	Type	Required	Default	Description
`agents`	`List[Union[Agent, Callable]]`	Yes	-	List of agent objects to get information about
`team_name`	`str`	No	None	Optional team name to display

Returns¶

str: Formatted string containing names and descriptions of all agents

Example¶

from swarms.structs.agent import Agent
from swarms.structs.multi_agent_exec import get_agents_info

agents = [
    Agent(
        agent_name="Research-Agent",
        system_prompt="You are a research assistant",
        model_name="gpt-4o-mini",
        max_loops=2,
        role="Researcher"
    ),
    Agent(
        agent_name="Analysis-Agent",
        system_prompt="You are a data analyst",
        model_name="gpt-4o-mini",
        max_loops=1,
        role="Analyst"
    )
]

info = get_agents_info(agents, team_name="Data Team")
print(info)
# Output:
# Available Agents for Team: Data Team
#
# [Agent 1]
# Name: Research-Agent
# Description: You are a research assistant
# Role: Researcher
# Model: gpt-4o-mini
# Max Loops: 2
#
# [Agent 2]
# Name: Analysis-Agent
# Description: You are a data analyst
# Role: Analyst
# Model: gpt-4o-mini
# Max Loops: 1

Complete Usage Examples¶

Advanced Multi-Agent Workflow Example¶

import asyncio
from swarms.structs.agent import Agent
from swarms.structs.multi_agent_exec import (
    run_agents_concurrently,
    run_agents_with_different_tasks,
    batched_grid_agent_execution,
    get_agents_info
)

# Create specialized agents
agents = [
    Agent(
        agent_name="Market-Researcher",
        system_prompt="You are a market research expert specializing in consumer behavior",
        model_name="gpt-4o-mini",
        max_loops=1,
        role="Researcher"
    ),
    Agent(
        agent_name="Data-Analyst",
        system_prompt="You are a data analyst expert in statistical analysis",
        model_name="gpt-4o-mini",
        max_loops=1,
        role="Analyst"
    ),
    Agent(
        agent_name="Strategy-Consultant",
        system_prompt="You are a strategy consultant specializing in business development",
        model_name="gpt-4o-mini",
        max_loops=1,
        role="Consultant"
    ),
    Agent(
        agent_name="Financial-Advisor",
        system_prompt="You are a financial advisor specializing in investment strategies",
        model_name="gpt-4o-mini",
        max_loops=1,
        role="Advisor"
    )
]

# Display agent information
print("=== Agent Information ===")
print(get_agents_info(agents, "Business Intelligence Team"))
print("\n" + "="*50 + "\n")

# Example 1: Same task for all agents
print("=== Example 1: Concurrent Execution with Same Task ===")
task = "Analyze the impact of remote work trends on commercial real estate market in 2024"
results = run_agents_concurrently(agents, task, max_workers=4)

for i, result in enumerate(results):
    print(f"\n{agents[i].agent_name} Analysis:")
    print(f"Result: {result}")

print("\n" + "="*50 + "\n")

# Example 2: Different tasks for different agents
print("=== Example 2: Different Tasks for Different Agents ===")
agent_task_pairs = [
    (agents[0], "Research consumer preferences for electric vehicles"),
    (agents[1], "Analyze sales data for EV market penetration"),
    (agents[2], "Develop marketing strategy for EV adoption"),
    (agents[3], "Assess financial viability of EV charging infrastructure")
]

results = run_agents_with_different_tasks(agent_task_pairs, batch_size=2)

for i, result in enumerate(results):
    agent, task = agent_task_pairs[i]
    print(f"\n{agent.agent_name} - Task: {task}")
    print(f"Result: {result}")

print("\n" + "="*50 + "\n")

# Example 3: Grid execution with matched agents and tasks
print("=== Example 3: Batched Grid Execution ===")
grid_agents = agents[:3]  # Use first 3 agents
grid_tasks = [
    "Forecast market trends for renewable energy",
    "Evaluate risk factors in green technology investments",
    "Compare traditional vs sustainable investment portfolios"
]

grid_results = batched_grid_agent_execution(grid_agents, grid_tasks, max_workers=3)

for i, result in enumerate(grid_results):
    print(f"\nTask {i+1}: {grid_tasks[i]}")
    print(f"Agent: {grid_agents[i].agent_name}")
    print(f"Result: {result}")

print("\n=== Workflow Complete ===")

Error Handling and Best Practices¶

from swarms.structs.agent import Agent
from swarms.structs.multi_agent_exec import run_agents_concurrently
import logging

# Set up logging
logging.basicConfig(level=logging.INFO)

# Create agents with error handling
agents = [
    Agent(
        agent_name=f"Agent-{i}",
        system_prompt="You are a helpful assistant",
        model_name="gpt-4o-mini",
        max_loops=1
    )
    for i in range(5)
]

task = "Perform a complex analysis task"

try:
    results = run_agents_concurrently(agents, task, max_workers=4)

    # Handle results (some may be exceptions)
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Agent {i+1} failed with error: {result}")
        else:
            print(f"Agent {i+1} succeeded: {result}")

except Exception as e:
    print(f"Execution failed: {e}")

# Best practices:
# 1. Always handle exceptions in results
# 2. Use appropriate max_workers based on system resources
# 3. Monitor memory usage for large agent counts
# 4. Consider batch processing for very large numbers of agents
# 5. Use uvloop functions for I/O intensive tasks

Performance Considerations¶

Technique	Best Use Case / Description
ThreadPoolExecutor	Best for CPU-bound tasks with moderate I/O
uvloop	Optimized for I/O-bound tasks, significantly faster than standard asyncio
Batch Processing	Prevents system overload with large numbers of agents
Resource Monitoring	Adjust worker counts based on system capabilities
Async/Await	Use async functions for better concurrency control

Multi-Agent Execution API Reference¶

Function Overview¶

Single Agent Functions¶

run_single_agent(agent, task, *args, **kwargs)¶

Signature¶

Parameters¶

Returns¶

Example¶

run_agent_async(agent, task)¶

Signature¶

Parameters¶

Returns¶

Example¶

Concurrent Execution Functions¶

run_agents_concurrently_async(agents, task)¶

Signature¶

Parameters¶

Returns¶

Example¶

run_agents_concurrently(agents, task, max_workers=None)¶

Signature¶

Parameters¶

Returns¶

Example¶

run_agents_concurrently_multiprocess(agents, task, batch_size=None)¶

Signature¶

Parameters¶

Returns¶

Example¶

Batched and Grid Execution¶

batched_grid_agent_execution(agents, tasks, max_workers=None)¶

Signature¶

Parameters¶

Returns¶

Raises¶

Example¶

run_agents_with_different_tasks(agent_task_pairs, batch_size=10, max_workers=None)¶

Signature¶

Parameters¶

Returns¶

Example¶

UVLoop Optimized Functions¶

run_agents_concurrently_uvloop(agents, task, max_workers=None)¶

Signature¶

Parameters¶

Returns¶

Raises¶

Example¶

run_agents_with_tasks_uvloop(agents, tasks, max_workers=None)¶

Signature¶

Parameters¶

Returns¶

Raises¶

Example¶

Utility Functions¶

get_swarms_info(swarms)¶

Signature¶

Parameters¶

Returns¶

Example¶

get_agents_info(agents, team_name=None)¶

Signature¶

Parameters¶

Returns¶

Example¶

Complete Usage Examples¶

Advanced Multi-Agent Workflow Example¶

Error Handling and Best Practices¶

Performance Considerations¶

`run_single_agent(agent, task, *args, **kwargs)`¶

`run_agent_async(agent, task)`¶

`run_agents_concurrently_async(agents, task)`¶

`run_agents_concurrently(agents, task, max_workers=None)`¶

`run_agents_concurrently_multiprocess(agents, task, batch_size=None)`¶

`batched_grid_agent_execution(agents, tasks, max_workers=None)`¶

`run_agents_with_different_tasks(agent_task_pairs, batch_size=10, max_workers=None)`¶

`run_agents_concurrently_uvloop(agents, task, max_workers=None)`¶

`run_agents_with_tasks_uvloop(agents, tasks, max_workers=None)`¶

`get_swarms_info(swarms)`¶

`get_agents_info(agents, team_name=None)`¶