ai-mirow-projects
/
swarms
mirror of https://github.com/kyegomez/swarms
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '79251be936'
${ noResults }
swarms/docs/swarms/structs/various_execution_methods.md

777 lines
23 KiB
Raw Blame History

# 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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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 |
Reference in new issue View Git Blame Copy Permalink