run_agents_concurrently_async,
run_single_agent,
run_agents_concurrently_multiprocess,
run_agents_sequentially,
run_agents_with_different_tasks,
run_agent_with_timeout,
run_agents_with_resource_monitoring,] + [CLEANUP]
pull/622/head
Your Name
1 year ago
parent
512ffe8a2f
commit
5ef4897f07
@ -0,0 +1,299 @@
|
||||
from loguru import logger
|
||||
|
||||
import os
|
||||
from typing import List
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
from swarm_models import OpenAIFunctionCaller, OpenAIChat
|
||||
|
||||
from swarms.structs.agent import Agent
|
||||
from swarms.structs.swarm_router import SwarmRouter
|
||||
|
||||
|
||||
class AgentConfig(BaseModel):
|
||||
"""Configuration for an individual agent in a swarm"""
|
||||
|
||||
name: str = Field(
|
||||
description="The name of the agent", example="Research-Agent"
|
||||
)
|
||||
description: str = Field(
|
||||
description="A description of the agent's purpose and capabilities",
|
||||
example="Agent responsible for researching and gathering information",
|
||||
)
|
||||
system_prompt: str = Field(
|
||||
description="The system prompt that defines the agent's behavior",
|
||||
example="You are a research agent. Your role is to gather and analyze information...",
|
||||
)
|
||||
max_loops: int = Field(
|
||||
description="Maximum number of reasoning loops the agent can perform",
|
||||
example=3,
|
||||
)
|
||||
|
||||
|
||||
class SwarmConfig(BaseModel):
|
||||
"""Configuration for a swarm of cooperative agents"""
|
||||
|
||||
name: str = Field(
|
||||
description="The name of the swarm",
|
||||
example="Research-Writing-Swarm",
|
||||
)
|
||||
description: str = Field(
|
||||
description="The description of the swarm's purpose and capabilities",
|
||||
example="A swarm of agents that work together to research topics and write articles",
|
||||
)
|
||||
agents: List[AgentConfig] = Field(
|
||||
description="The list of agents that make up the swarm",
|
||||
example=[
|
||||
AgentConfig(
|
||||
name="Research-Agent",
|
||||
description="Gathers information",
|
||||
system_prompt="You are a research agent...",
|
||||
max_loops=2,
|
||||
),
|
||||
AgentConfig(
|
||||
name="Writing-Agent",
|
||||
description="Writes content",
|
||||
system_prompt="You are a writing agent...",
|
||||
max_loops=1,
|
||||
),
|
||||
],
|
||||
)
|
||||
max_loops: int = Field(
|
||||
description="The maximum number of loops to run the swarm",
|
||||
example=1,
|
||||
)
|
||||
|
||||
|
||||
# Get the OpenAI API key from the environment variable
|
||||
api_key = os.getenv("OPENAI_API_KEY")
|
||||
|
||||
# Create an instance of the OpenAIChat class
|
||||
model = OpenAIChat(
|
||||
openai_api_key=api_key, model_name="gpt-4o-mini", temperature=0.1
|
||||
)
|
||||
|
||||
|
||||
BOSS_SYSTEM_PROMPT = """
|
||||
Manage a swarm of worker agents to efficiently serve the user by deciding whether to create new agents or delegate tasks. Ensure operations are efficient and effective.
|
||||
|
||||
### Instructions:
|
||||
|
||||
1. **Task Assignment**:
|
||||
- Analyze available worker agents when a task is presented.
|
||||
- Delegate tasks to existing agents with clear, direct, and actionable instructions if an appropriate agent is available.
|
||||
- If no suitable agent exists, create a new agent with a fitting system prompt to handle the task.
|
||||
|
||||
2. **Agent Creation**:
|
||||
- Name agents according to the task they are intended to perform (e.g., "Twitter Marketing Agent").
|
||||
- Provide each new agent with a concise and clear system prompt that includes its role, objectives, and any tools it can utilize.
|
||||
|
||||
3. **Efficiency**:
|
||||
- Minimize redundancy and maximize task completion speed.
|
||||
- Avoid unnecessary agent creation if an existing agent can fulfill the task.
|
||||
|
||||
4. **Communication**:
|
||||
- Be explicit in task delegation instructions to avoid ambiguity and ensure effective task execution.
|
||||
- Require agents to report back on task completion or encountered issues.
|
||||
|
||||
5. **Reasoning and Decisions**:
|
||||
- Offer brief reasoning when selecting or creating agents to maintain transparency.
|
||||
- Avoid using an agent if unnecessary, with a clear explanation if no agents are suitable for a task.
|
||||
|
||||
# Output Format
|
||||
|
||||
Present your plan in clear, bullet-point format or short concise paragraphs, outlining task assignment, agent creation, efficiency strategies, and communication protocols.
|
||||
|
||||
# Notes
|
||||
|
||||
- Preserve transparency by always providing reasoning for task-agent assignments and creation.
|
||||
- Ensure instructions to agents are unambiguous to minimize error.
|
||||
|
||||
"""
|
||||
|
||||
|
||||
class AutoSwarmBuilder:
|
||||
"""A class that automatically builds and manages swarms of AI agents.
|
||||
|
||||
This class handles the creation, coordination and execution of multiple AI agents working
|
||||
together as a swarm to accomplish complex tasks. It uses a boss agent to delegate work
|
||||
and create new specialized agents as needed.
|
||||
|
||||
Args:
|
||||
name (str): The name of the swarm
|
||||
description (str): A description of the swarm's purpose
|
||||
verbose (bool, optional): Whether to output detailed logs. Defaults to True.
|
||||
max_loops (int, optional): Maximum number of execution loops. Defaults to 1.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
name: str = None,
|
||||
description: str = None,
|
||||
verbose: bool = True,
|
||||
max_loops: int = 1,
|
||||
):
|
||||
self.name = name
|
||||
self.description = description
|
||||
self.verbose = verbose
|
||||
self.max_loops = max_loops
|
||||
self.agents_pool = []
|
||||
logger.info(
|
||||
f"Initialized AutoSwarmBuilder: {name} {description}"
|
||||
)
|
||||
|
||||
# @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
|
||||
def run(self, task: str, image_url: str = None, *args, **kwargs):
|
||||
"""Run the swarm on a given task.
|
||||
|
||||
Args:
|
||||
task (str): The task to be accomplished
|
||||
image_url (str, optional): URL of an image input if needed. Defaults to None.
|
||||
*args: Variable length argument list
|
||||
**kwargs: Arbitrary keyword arguments
|
||||
|
||||
Returns:
|
||||
The output from the swarm's execution
|
||||
"""
|
||||
logger.info(f"Running swarm on task: {task}")
|
||||
agents = self._create_agents(task, image_url, *args, **kwargs)
|
||||
logger.info(f"Agents created {len(agents)}")
|
||||
logger.info("Routing task through swarm")
|
||||
output = self.swarm_router(agents, task, image_url)
|
||||
logger.info(f"Swarm execution complete with output: {output}")
|
||||
return output
|
||||
|
||||
# @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
|
||||
def _create_agents(self, task: str, *args, **kwargs):
|
||||
"""Create the necessary agents for a task.
|
||||
|
||||
Args:
|
||||
task (str): The task to create agents for
|
||||
*args: Variable length argument list
|
||||
**kwargs: Arbitrary keyword arguments
|
||||
|
||||
Returns:
|
||||
list: List of created agents
|
||||
"""
|
||||
logger.info("Creating agents for task")
|
||||
model = OpenAIFunctionCaller(
|
||||
system_prompt=BOSS_SYSTEM_PROMPT,
|
||||
api_key=os.getenv("OPENAI_API_KEY"),
|
||||
temperature=0.1,
|
||||
base_model=SwarmConfig,
|
||||
)
|
||||
|
||||
agents_dictionary = model.run(task)
|
||||
logger.info(f"Agents dictionary: {agents_dictionary}")
|
||||
|
||||
# Convert dictionary to SwarmConfig if needed
|
||||
if isinstance(agents_dictionary, dict):
|
||||
agents_dictionary = SwarmConfig(**agents_dictionary)
|
||||
|
||||
# Set swarm config
|
||||
self.name = agents_dictionary.name
|
||||
self.description = agents_dictionary.description
|
||||
self.max_loops = getattr(
|
||||
agents_dictionary, "max_loops", 1
|
||||
) # Default to 1 if not set
|
||||
|
||||
logger.info(
|
||||
f"Swarm config: {self.name}, {self.description}, {self.max_loops}"
|
||||
)
|
||||
|
||||
# Create agents from config
|
||||
agents = []
|
||||
for agent_config in agents_dictionary.agents:
|
||||
# Convert dict to AgentConfig if needed
|
||||
if isinstance(agent_config, dict):
|
||||
agent_config = AgentConfig(**agent_config)
|
||||
|
||||
agent = self.build_agent(
|
||||
agent_name=agent_config.name,
|
||||
agent_description=agent_config.description,
|
||||
agent_system_prompt=agent_config.system_prompt,
|
||||
max_loops=agent_config.max_loops,
|
||||
)
|
||||
agents.append(agent)
|
||||
|
||||
return agents
|
||||
|
||||
def build_agent(
|
||||
self,
|
||||
agent_name: str,
|
||||
agent_description: str,
|
||||
agent_system_prompt: str,
|
||||
max_loops: int = 1,
|
||||
):
|
||||
"""Build a single agent with the given specifications.
|
||||
|
||||
Args:
|
||||
agent_name (str): Name of the agent
|
||||
agent_description (str): Description of the agent's purpose
|
||||
agent_system_prompt (str): The system prompt for the agent
|
||||
|
||||
Returns:
|
||||
Agent: The constructed agent instance
|
||||
"""
|
||||
logger.info(f"Building agent: {agent_name}")
|
||||
agent = Agent(
|
||||
agent_name=agent_name,
|
||||
description=agent_description,
|
||||
system_prompt=agent_system_prompt,
|
||||
llm=model,
|
||||
max_loops=max_loops,
|
||||
autosave=True,
|
||||
dashboard=False,
|
||||
verbose=True,
|
||||
dynamic_temperature_enabled=True,
|
||||
saved_state_path=f"{agent_name}.json",
|
||||
user_name="swarms_corp",
|
||||
retry_attempts=1,
|
||||
context_length=200000,
|
||||
return_step_meta=False,
|
||||
output_type="str", # "json", "dict", "csv" OR "string" soon "yaml" and
|
||||
streaming_on=False,
|
||||
auto_generate_prompt=True,
|
||||
)
|
||||
|
||||
return agent
|
||||
|
||||
# @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
|
||||
def swarm_router(
|
||||
self,
|
||||
agents: List[Agent],
|
||||
task: str,
|
||||
image_url: str = None,
|
||||
*args,
|
||||
**kwargs,
|
||||
):
|
||||
"""Route tasks between agents in the swarm.
|
||||
|
||||
Args:
|
||||
agents (List[Agent]): List of available agents
|
||||
task (str): The task to route
|
||||
image_url (str, optional): URL of an image input if needed. Defaults to None.
|
||||
*args: Variable length argument list
|
||||
**kwargs: Arbitrary keyword arguments
|
||||
|
||||
Returns:
|
||||
The output from the routed task execution
|
||||
"""
|
||||
logger.info("Routing task through swarm")
|
||||
swarm_router_instance = SwarmRouter(
|
||||
agents=agents,
|
||||
swarm_type="auto",
|
||||
max_loops=1,
|
||||
)
|
||||
|
||||
return swarm_router_instance.run(
|
||||
self.name + " " + self.description + " " + task,
|
||||
)
|
||||
|
||||
|
||||
example = AutoSwarmBuilder()
|
||||
|
||||
print(
|
||||
example.run(
|
||||
"Write multiple blog posts about the latest advancements in swarm intelligence all at once"
|
||||
)
|
||||
)
|
||||
@ -0,0 +1,162 @@
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
from swarms import Agent
|
||||
from swarm_models import OpenAIChat
|
||||
from swarms.structs.swarm_router import SwarmRouter
|
||||
|
||||
load_dotenv()
|
||||
|
||||
# Get the OpenAI API key from the environment variable
|
||||
api_key = os.getenv("GROQ_API_KEY")
|
||||
|
||||
# Model
|
||||
model = OpenAIChat(
|
||||
openai_api_base="https://api.groq.com/openai/v1",
|
||||
openai_api_key=api_key,
|
||||
model_name="llama-3.1-70b-versatile",
|
||||
temperature=0.1,
|
||||
)
|
||||
# Define specialized system prompts for each agent
|
||||
DATA_EXTRACTOR_PROMPT = """You are a highly specialized private equity agent focused on data extraction from various documents. Your expertise includes:
|
||||
1. Extracting key financial metrics (revenue, EBITDA, growth rates, etc.) from financial statements and reports
|
||||
2. Identifying and extracting important contract terms from legal documents
|
||||
3. Pulling out relevant market data from industry reports and analyses
|
||||
4. Extracting operational KPIs from management presentations and internal reports
|
||||
5. Identifying and extracting key personnel information from organizational charts and bios
|
||||
Provide accurate, structured data extracted from various document types to support investment analysis."""
|
||||
|
||||
SUMMARIZER_PROMPT = """You are an expert private equity agent specializing in summarizing complex documents. Your core competencies include:
|
||||
1. Distilling lengthy financial reports into concise executive summaries
|
||||
2. Summarizing legal documents, highlighting key terms and potential risks
|
||||
3. Condensing industry reports to capture essential market trends and competitive dynamics
|
||||
4. Summarizing management presentations to highlight key strategic initiatives and projections
|
||||
5. Creating brief overviews of technical documents, emphasizing critical points for non-technical stakeholders
|
||||
Deliver clear, concise summaries that capture the essence of various documents while highlighting information crucial for investment decisions."""
|
||||
|
||||
FINANCIAL_ANALYST_PROMPT = """You are a specialized private equity agent focused on financial analysis. Your key responsibilities include:
|
||||
1. Analyzing historical financial statements to identify trends and potential issues
|
||||
2. Evaluating the quality of earnings and potential adjustments to EBITDA
|
||||
3. Assessing working capital requirements and cash flow dynamics
|
||||
4. Analyzing capital structure and debt capacity
|
||||
5. Evaluating financial projections and underlying assumptions
|
||||
Provide thorough, insightful financial analysis to inform investment decisions and valuation."""
|
||||
|
||||
MARKET_ANALYST_PROMPT = """You are a highly skilled private equity agent specializing in market analysis. Your expertise covers:
|
||||
1. Analyzing industry trends, growth drivers, and potential disruptors
|
||||
2. Evaluating competitive landscape and market positioning
|
||||
3. Assessing market size, segmentation, and growth potential
|
||||
4. Analyzing customer dynamics, including concentration and loyalty
|
||||
5. Identifying potential regulatory or macroeconomic impacts on the market
|
||||
Deliver comprehensive market analysis to assess the attractiveness and risks of potential investments."""
|
||||
|
||||
OPERATIONAL_ANALYST_PROMPT = """You are an expert private equity agent focused on operational analysis. Your core competencies include:
|
||||
1. Evaluating operational efficiency and identifying improvement opportunities
|
||||
2. Analyzing supply chain and procurement processes
|
||||
3. Assessing sales and marketing effectiveness
|
||||
4. Evaluating IT systems and digital capabilities
|
||||
5. Identifying potential synergies in merger or add-on acquisition scenarios
|
||||
Provide detailed operational analysis to uncover value creation opportunities and potential risks."""
|
||||
|
||||
# Initialize specialized agents
|
||||
data_extractor_agent = Agent(
|
||||
agent_name="Data-Extractor",
|
||||
system_prompt=DATA_EXTRACTOR_PROMPT,
|
||||
llm=model,
|
||||
max_loops=1,
|
||||
autosave=True,
|
||||
verbose=True,
|
||||
dynamic_temperature_enabled=True,
|
||||
saved_state_path="data_extractor_agent.json",
|
||||
user_name="pe_firm",
|
||||
retry_attempts=1,
|
||||
context_length=200000,
|
||||
output_type="string",
|
||||
)
|
||||
|
||||
summarizer_agent = Agent(
|
||||
agent_name="Document-Summarizer",
|
||||
system_prompt=SUMMARIZER_PROMPT,
|
||||
llm=model,
|
||||
max_loops=1,
|
||||
autosave=True,
|
||||
verbose=True,
|
||||
dynamic_temperature_enabled=True,
|
||||
saved_state_path="summarizer_agent.json",
|
||||
user_name="pe_firm",
|
||||
retry_attempts=1,
|
||||
context_length=200000,
|
||||
output_type="string",
|
||||
)
|
||||
|
||||
financial_analyst_agent = Agent(
|
||||
agent_name="Financial-Analyst",
|
||||
system_prompt=FINANCIAL_ANALYST_PROMPT,
|
||||
llm=model,
|
||||
max_loops=1,
|
||||
autosave=True,
|
||||
verbose=True,
|
||||
dynamic_temperature_enabled=True,
|
||||
saved_state_path="financial_analyst_agent.json",
|
||||
user_name="pe_firm",
|
||||
retry_attempts=1,
|
||||
context_length=200000,
|
||||
output_type="string",
|
||||
)
|
||||
|
||||
market_analyst_agent = Agent(
|
||||
agent_name="Market-Analyst",
|
||||
system_prompt=MARKET_ANALYST_PROMPT,
|
||||
llm=model,
|
||||
max_loops=1,
|
||||
autosave=True,
|
||||
verbose=True,
|
||||
dynamic_temperature_enabled=True,
|
||||
saved_state_path="market_analyst_agent.json",
|
||||
user_name="pe_firm",
|
||||
retry_attempts=1,
|
||||
context_length=200000,
|
||||
output_type="string",
|
||||
)
|
||||
|
||||
operational_analyst_agent = Agent(
|
||||
agent_name="Operational-Analyst",
|
||||
system_prompt=OPERATIONAL_ANALYST_PROMPT,
|
||||
llm=model,
|
||||
max_loops=1,
|
||||
autosave=True,
|
||||
verbose=True,
|
||||
dynamic_temperature_enabled=True,
|
||||
saved_state_path="operational_analyst_agent.json",
|
||||
user_name="pe_firm",
|
||||
retry_attempts=1,
|
||||
context_length=200000,
|
||||
output_type="string",
|
||||
)
|
||||
|
||||
# Initialize the SwarmRouter
|
||||
router = SwarmRouter(
|
||||
name="pe-document-analysis-swarm",
|
||||
description="Analyze documents for private equity due diligence and investment decision-making",
|
||||
max_loops=1,
|
||||
agents=[
|
||||
data_extractor_agent,
|
||||
summarizer_agent,
|
||||
# financial_analyst_agent,
|
||||
# market_analyst_agent,
|
||||
# operational_analyst_agent,
|
||||
],
|
||||
swarm_type="auto", # or "SequentialWorkflow" or "ConcurrentWorkflow" or
|
||||
auto_generate_prompts=True,
|
||||
)
|
||||
|
||||
# Example usage
|
||||
if __name__ == "__main__":
|
||||
# Run a comprehensive private equity document analysis task
|
||||
result = router.run(
|
||||
"Where is the best place to find template term sheets for series A startups. Provide links and references"
|
||||
)
|
||||
print(result)
|
||||
|
||||
# Retrieve and print logs
|
||||
for log in router.get_logs():
|
||||
print(f"{log.timestamp} - {log.level}: {log.message}")
|
||||
@ -0,0 +1,168 @@
|
||||
# Concurrent Agents API Reference
|
||||
|
||||
This documentation covers the API for running multiple agents concurrently using various execution strategies. The implementation uses `asyncio` with `uvloop` for enhanced performance and `ThreadPoolExecutor` for handling CPU-bound operations.
|
||||
|
||||
## Table of Contents
|
||||
- [Core Functions](#core-functions)
|
||||
- [Advanced Functions](#advanced-functions)
|
||||
- [Utility Functions](#utility-functions)
|
||||
- [Resource Monitoring](#resource-monitoring)
|
||||
- [Usage Examples](#usage-examples)
|
||||
|
||||
## Core Functions
|
||||
|
||||
### run_agents_concurrently()
|
||||
|
||||
Primary function for running multiple agents concurrently with optimized performance using both uvloop and ThreadPoolExecutor.
|
||||
|
||||
#### Arguments
|
||||
|
||||
| Parameter | Type | Required | Default | Description |
|
||||
|-------------|----------------|----------|----------------|-------------|
|
||||
| agents | List[AgentType]| Yes | - | List of Agent instances to run concurrently |
|
||||
| task | str | Yes | - | Task string to execute |
|
||||
| batch_size | int | No | CPU count | Number of agents to run in parallel in each batch |
|
||||
| max_workers | int | No | CPU count * 2 | Maximum number of threads in the executor |
|
||||
|
||||
#### Returns
|
||||
`List[Any]`: List of outputs from each agent
|
||||
|
||||
#### Flow Diagram
|
||||
|
||||
```mermaid
|
||||
graph TD
|
||||
A[Start] --> B[Initialize ThreadPoolExecutor]
|
||||
B --> C[Split Agents into Batches]
|
||||
C --> D[Process Batch]
|
||||
D --> E{More Batches?}
|
||||
E -->|Yes| D
|
||||
E -->|No| F[Combine Results]
|
||||
F --> G[Return Results]
|
||||
|
||||
subgraph "Batch Processing"
|
||||
D --> H[Run Agents Async]
|
||||
H --> I[Wait for Completion]
|
||||
I --> J[Collect Batch Results]
|
||||
end
|
||||
```
|
||||
|
||||
### run_agents_sequentially()
|
||||
|
||||
Runs multiple agents sequentially for baseline comparison or simple use cases.
|
||||
|
||||
#### Arguments
|
||||
|
||||
| Parameter | Type | Required | Default | Description |
|
||||
|-----------|----------------|----------|---------|-------------|
|
||||
| agents | List[AgentType]| Yes | - | List of Agent instances to run |
|
||||
| task | str | Yes | - | Task string to execute |
|
||||
|
||||
#### Returns
|
||||
`List[Any]`: List of outputs from each agent
|
||||
|
||||
## Advanced Functions
|
||||
|
||||
### run_agents_with_different_tasks()
|
||||
|
||||
Runs multiple agents with different tasks concurrently.
|
||||
|
||||
#### Arguments
|
||||
|
||||
| Parameter | Type | Required | Default | Description |
|
||||
|-----------------|-------------------------------|----------|----------------|-------------|
|
||||
| agent_task_pairs| List[tuple[AgentType, str]] | Yes | - | List of (agent, task) tuples |
|
||||
| batch_size | int | No | CPU count | Number of agents to run in parallel |
|
||||
| max_workers | int | No | CPU count * 2 | Maximum number of threads |
|
||||
|
||||
### run_agents_with_timeout()
|
||||
|
||||
Runs multiple agents concurrently with timeout limits.
|
||||
|
||||
#### Arguments
|
||||
|
||||
| Parameter | Type | Required | Default | Description |
|
||||
|-------------|----------------|----------|----------------|-------------|
|
||||
| agents | List[AgentType]| Yes | - | List of Agent instances |
|
||||
| task | str | Yes | - | Task string to execute |
|
||||
| timeout | float | Yes | - | Timeout in seconds for each agent |
|
||||
| batch_size | int | No | CPU count | Number of agents to run in parallel |
|
||||
| max_workers | int | No | CPU count * 2 | Maximum number of threads |
|
||||
|
||||
## Usage Examples
|
||||
|
||||
```python
|
||||
from swarms.structs.agent import Agent
|
||||
from your_module import run_agents_concurrently
|
||||
|
||||
# Initialize agents
|
||||
agents = [
|
||||
Agent(
|
||||
agent_name=f"Analysis-Agent-{i}",
|
||||
system_prompt="You are a financial analysis expert",
|
||||
llm=model,
|
||||
max_loops=1
|
||||
)
|
||||
for i in range(5)
|
||||
]
|
||||
|
||||
# Basic concurrent execution
|
||||
task = "Analyze the impact of rising interest rates on tech stocks"
|
||||
outputs = run_agents_concurrently(agents, task)
|
||||
|
||||
# Running with timeout
|
||||
outputs_with_timeout = run_agents_with_timeout(
|
||||
agents=agents,
|
||||
task=task,
|
||||
timeout=30.0,
|
||||
batch_size=2
|
||||
)
|
||||
|
||||
# Running different tasks
|
||||
task_pairs = [
|
||||
(agents[0], "Analyze tech stocks"),
|
||||
(agents[1], "Analyze energy stocks"),
|
||||
(agents[2], "Analyze retail stocks")
|
||||
]
|
||||
different_outputs = run_agents_with_different_tasks(task_pairs)
|
||||
```
|
||||
|
||||
## Resource Monitoring
|
||||
|
||||
### ResourceMetrics
|
||||
|
||||
A dataclass for system resource metrics.
|
||||
|
||||
#### Properties
|
||||
|
||||
| Property | Type | Description |
|
||||
|----------------|-------|-------------|
|
||||
| cpu_percent | float | Current CPU usage percentage |
|
||||
| memory_percent | float | Current memory usage percentage |
|
||||
| active_threads | int | Number of active threads |
|
||||
|
||||
### run_agents_with_resource_monitoring()
|
||||
|
||||
Runs agents with system resource monitoring and adaptive batch sizing.
|
||||
|
||||
#### Arguments
|
||||
|
||||
| Parameter | Type | Required | Default | Description |
|
||||
|------------------|----------------|----------|---------|-------------|
|
||||
| agents | List[AgentType]| Yes | - | List of Agent instances |
|
||||
| task | str | Yes | - | Task string to execute |
|
||||
| cpu_threshold | float | No | 90.0 | Max CPU usage percentage |
|
||||
| memory_threshold | float | No | 90.0 | Max memory usage percentage |
|
||||
| check_interval | float | No | 1.0 | Resource check interval in seconds |
|
||||
|
||||
## Performance Considerations
|
||||
|
||||
- All functions are decorated with `@profile_func` for performance monitoring
|
||||
- Default batch sizes and worker counts are optimized based on CPU cores
|
||||
- Resource monitoring helps prevent system overload
|
||||
- Using `uvloop` provides better performance than standard `asyncio`
|
||||
|
||||
## Error Handling
|
||||
|
||||
- Functions handle asyncio event loop creation/retrieval
|
||||
- Timeout mechanism prevents infinite waiting
|
||||
- Resource monitoring allows for adaptive performance adjustment
|
||||
@ -1,84 +0,0 @@
|
||||
import os
|
||||
from swarms import Agent
|
||||
from swarm_models import OpenAIChat
|
||||
from swarms.prompts.prompt_generator_optimizer import (
|
||||
prompt_generator_sys_prompt,
|
||||
)
|
||||
from dotenv import load_dotenv
|
||||
from swarms.agents.prompt_generator_agent import PromptGeneratorAgent
|
||||
from yaml import dump
|
||||
|
||||
load_dotenv()
|
||||
|
||||
|
||||
def generate_prompt(
|
||||
num_loops: int = 1,
|
||||
autosave: bool = True,
|
||||
save_to_yaml: bool = False,
|
||||
prompt: str = None,
|
||||
save_format: str = "yaml",
|
||||
) -> None:
|
||||
"""
|
||||
This function creates and runs a prompt generator agent with default settings for number of loops and autosave.
|
||||
|
||||
Args:
|
||||
num_loops (int, optional): The number of loops to run the agent. Defaults to 1.
|
||||
autosave (bool, optional): Whether to autosave the agent's state. Defaults to True.
|
||||
save_to_yaml (bool, optional): Whether to save the agent's configuration to a YAML file. Defaults to False.
|
||||
prompt (str): The prompt to generate.
|
||||
save_format (str, optional): The format in which to save the generated prompt. Defaults to "yaml".
|
||||
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
# Get the OpenAI API key from the environment variable
|
||||
api_key = os.getenv("OPENAI_API_KEY")
|
||||
|
||||
# Create an instance of the OpenAIChat class
|
||||
model = OpenAIChat(
|
||||
openai_api_key=api_key,
|
||||
model_name="gpt-4o-mini",
|
||||
temperature=0.1,
|
||||
max_tokens=2000,
|
||||
)
|
||||
|
||||
# Initialize the agent
|
||||
agent = Agent(
|
||||
agent_name="Prompt-Optimizer",
|
||||
system_prompt=prompt_generator_sys_prompt.get_prompt(),
|
||||
llm=model,
|
||||
max_loops=num_loops,
|
||||
autosave=autosave,
|
||||
dashboard=False,
|
||||
verbose=True,
|
||||
dynamic_temperature_enabled=True,
|
||||
saved_state_path="optimizer_agent.json",
|
||||
user_name="swarms_corp",
|
||||
retry_attempts=1,
|
||||
context_length=200000,
|
||||
return_step_meta=False,
|
||||
output_type="string",
|
||||
)
|
||||
|
||||
# Main Class
|
||||
prompt_generator = PromptGeneratorAgent(agent)
|
||||
|
||||
# Run the agent
|
||||
prompt_generator.run(prompt, save_format)
|
||||
|
||||
if save_to_yaml:
|
||||
with open("agent_config.yaml", "w") as file:
|
||||
dump(agent.config, file)
|
||||
|
||||
|
||||
# # Example usage
|
||||
# if __name__ == "__main__":
|
||||
# try:
|
||||
# create_and_run_agent(
|
||||
# num_loops=1,
|
||||
# autosave=True,
|
||||
# save_to_yaml=True,
|
||||
# prompt="Generate an amazing prompt for analyzing healthcare insurance documents",
|
||||
# )
|
||||
# except Exception as e:
|
||||
# logger.error(f"An error occurred: {e}")
|
||||
@ -1,92 +0,0 @@
|
||||
import json
|
||||
import time
|
||||
import uuid
|
||||
|
||||
import yaml
|
||||
from dotenv import load_dotenv
|
||||
from loguru import logger
|
||||
|
||||
from swarms.structs.agent import Agent
|
||||
from swarms.prompts.prompt_generator_optimizer import (
|
||||
prompt_generator_sys_prompt,
|
||||
)
|
||||
|
||||
|
||||
load_dotenv()
|
||||
|
||||
|
||||
class PromptGeneratorAgent:
|
||||
"""
|
||||
A class representing a prompt generator agent.
|
||||
|
||||
Attributes:
|
||||
----------
|
||||
agent : Agent
|
||||
The underlying agent instance.
|
||||
"""
|
||||
|
||||
def __init__(self, agent: Agent):
|
||||
"""
|
||||
Initializes the PromptGeneratorAgent instance.
|
||||
|
||||
Args:
|
||||
----
|
||||
agent : Agent
|
||||
The agent instance to be used for prompt generation.
|
||||
"""
|
||||
self.agent = agent
|
||||
|
||||
def run(self, task: str, format: str = "json") -> str:
|
||||
"""
|
||||
Runs the prompt generator agent with the given task description and saves the generated prompt with the given metadata in the specified format.
|
||||
|
||||
Args:
|
||||
----
|
||||
task : str
|
||||
The task description to be used for prompt generation.
|
||||
metadata : Dict[str, Any]
|
||||
The metadata to be saved along with the prompt.
|
||||
format : str, optional
|
||||
The format in which the prompt should be saved (default is "json").
|
||||
|
||||
Returns:
|
||||
-------
|
||||
str
|
||||
The generated prompt.
|
||||
"""
|
||||
prompt = self.agent.run(task)
|
||||
self.save_prompt(prompt, format)
|
||||
return prompt
|
||||
|
||||
def save_prompt(
|
||||
self,
|
||||
prompt: str,
|
||||
format: str = "yaml",
|
||||
):
|
||||
"""
|
||||
Saves the generated prompt with the given metadata in the specified format using the prompt generator sys prompt model dump.
|
||||
|
||||
Args:
|
||||
----
|
||||
prompt : str
|
||||
The generated prompt to be saved.
|
||||
metadata : Dict[str, Any]
|
||||
The metadata to be saved along with the prompt.
|
||||
format : str, optional
|
||||
The format in which the prompt should be saved (default is "json").
|
||||
"""
|
||||
data = {
|
||||
"prompt_history": prompt_generator_sys_prompt.model_dump(),
|
||||
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"prompt": prompt,
|
||||
}
|
||||
if format == "json":
|
||||
with open(f"prompt_{uuid.uuid4()}.json", "w") as f:
|
||||
json.dump(data, f, indent=4)
|
||||
elif format == "yaml":
|
||||
with open(f"prompt_{uuid.uuid4()}.yaml", "w") as f:
|
||||
yaml.dump(data, f)
|
||||
else:
|
||||
logger.error(
|
||||
"Invalid format. Only 'json' and 'yaml' are supported."
|
||||
)
|
||||
@ -1,20 +0,0 @@
|
||||
from typing import List, Any, Dict, Optional
|
||||
|
||||
|
||||
class AutoSwarmBuilder:
|
||||
def __init__(self, task: str, num_agents: int = 10, batch_size: int = 10):
|
||||
self.task = task
|
||||
self.num_agents = num_agents
|
||||
self.batch_size = batch_size
|
||||
|
||||
def run(self, task: str, image_url: str = None, *args, **kwargs):
|
||||
pass
|
||||
|
||||
def _create_swarm(self):
|
||||
pass
|
||||
|
||||
def _create_agents(self):
|
||||
pass
|
||||
|
||||
def _run_agents(self):
|
||||
pass
|
||||
@ -0,0 +1,344 @@
|
||||
import asyncio
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
import psutil
|
||||
from dataclasses import dataclass
|
||||
import threading
|
||||
from typing import List, Union, Any, Callable
|
||||
from multiprocessing import cpu_count
|
||||
|
||||
import uvloop
|
||||
from swarms.structs.agent import Agent
|
||||
from swarms.utils.calculate_func_metrics import profile_func
|
||||
|
||||
# Use uvloop for faster asyncio event loop
|
||||
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
|
||||
|
||||
# Type definitions
|
||||
AgentType = Union[Agent, Callable]
|
||||
|
||||
|
||||
def run_single_agent(agent: AgentType, task: str) -> Any:
|
||||
"""Run a single agent synchronously"""
|
||||
return agent.run(task)
|
||||
|
||||
|
||||
async def run_agent_async(
|
||||
agent: AgentType, task: str, executor: ThreadPoolExecutor
|
||||
) -> Any:
|
||||
"""
|
||||
Run an agent asynchronously using a thread executor.
|
||||
|
||||
Args:
|
||||
agent: Agent instance to run
|
||||
task: Task string to execute
|
||||
executor: ThreadPoolExecutor instance for handling CPU-bound operations
|
||||
|
||||
Returns:
|
||||
Agent execution result
|
||||
"""
|
||||
loop = asyncio.get_event_loop()
|
||||
return await loop.run_in_executor(
|
||||
executor, run_single_agent, agent, task
|
||||
)
|
||||
|
||||
|
||||
async def run_agents_concurrently_async(
|
||||
agents: List[AgentType], task: str, executor: ThreadPoolExecutor
|
||||
) -> List[Any]:
|
||||
"""
|
||||
Run multiple agents concurrently using asyncio and thread executor.
|
||||
|
||||
Args:
|
||||
agents: List of Agent instances to run concurrently
|
||||
task: Task string to execute
|
||||
executor: ThreadPoolExecutor for CPU-bound operations
|
||||
|
||||
Returns:
|
||||
List of outputs from each agent
|
||||
"""
|
||||
results = await asyncio.gather(
|
||||
*(run_agent_async(agent, task, executor) for agent in agents)
|
||||
)
|
||||
return results
|
||||
|
||||
|
||||
@profile_func
|
||||
def run_agents_concurrently(
|
||||
agents: List[AgentType],
|
||||
task: str,
|
||||
batch_size: int = None,
|
||||
max_workers: int = None,
|
||||
) -> List[Any]:
|
||||
"""
|
||||
Optimized concurrent agent runner using both uvloop and ThreadPoolExecutor.
|
||||
|
||||
Args:
|
||||
agents: List of Agent instances to run concurrently
|
||||
task: Task string to execute
|
||||
batch_size: Number of agents to run in parallel in each batch (defaults to CPU count)
|
||||
max_workers: Maximum number of threads in the executor (defaults to CPU count * 2)
|
||||
|
||||
Returns:
|
||||
List of outputs from each agent
|
||||
"""
|
||||
# Optimize defaults based on system resources
|
||||
cpu_cores = cpu_count()
|
||||
batch_size = batch_size or cpu_cores
|
||||
max_workers = max_workers or cpu_cores * 2
|
||||
|
||||
results = []
|
||||
|
||||
# Get or create event loop
|
||||
try:
|
||||
loop = asyncio.get_event_loop()
|
||||
except RuntimeError:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
|
||||
# Create a shared thread pool executor with optimal worker count
|
||||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||||
# Process agents in batches
|
||||
for i in range(0, len(agents), batch_size):
|
||||
batch = agents[i : i + batch_size]
|
||||
batch_results = loop.run_until_complete(
|
||||
run_agents_concurrently_async(batch, task, executor)
|
||||
)
|
||||
results.extend(batch_results)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
@profile_func
|
||||
def run_agents_concurrently_multiprocess(
|
||||
agents: List[Agent], task: str, batch_size: int = cpu_count()
|
||||
) -> List[Any]:
|
||||
"""
|
||||
Manage and run multiple agents concurrently in batches, with optimized performance.
|
||||
|
||||
Args:
|
||||
agents (List[Agent]): List of Agent instances to run concurrently.
|
||||
task (str): The task string to execute by all agents.
|
||||
batch_size (int, optional): Number of agents to run in parallel in each batch.
|
||||
Defaults to the number of CPU cores.
|
||||
|
||||
Returns:
|
||||
List[Any]: A list of outputs from each agent.
|
||||
"""
|
||||
results = []
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
# Process agents in batches to avoid overwhelming system resources
|
||||
for i in range(0, len(agents), batch_size):
|
||||
batch = agents[i : i + batch_size]
|
||||
batch_results = loop.run_until_complete(
|
||||
run_agents_concurrently_async(batch, task)
|
||||
)
|
||||
results.extend(batch_results)
|
||||
|
||||
return results
|
||||
|
||||
@profile_func
|
||||
def run_agents_sequentially(agents: List[AgentType], task: str) -> List[Any]:
|
||||
"""
|
||||
Run multiple agents sequentially for baseline comparison.
|
||||
|
||||
Args:
|
||||
agents: List of Agent instances to run
|
||||
task: Task string to execute
|
||||
|
||||
Returns:
|
||||
List of outputs from each agent
|
||||
"""
|
||||
return [run_single_agent(agent, task) for agent in agents]
|
||||
|
||||
|
||||
@profile_func
|
||||
def run_agents_with_different_tasks(
|
||||
agent_task_pairs: List[tuple[AgentType, str]],
|
||||
batch_size: int = None,
|
||||
max_workers: int = None
|
||||
) -> List[Any]:
|
||||
"""
|
||||
Run multiple agents with different tasks concurrently.
|
||||
|
||||
Args:
|
||||
agent_task_pairs: List of (agent, task) tuples
|
||||
batch_size: Number of agents to run in parallel
|
||||
max_workers: Maximum number of threads
|
||||
|
||||
Returns:
|
||||
List of outputs from each agent
|
||||
"""
|
||||
async def run_pair_async(pair: tuple[AgentType, str], executor: ThreadPoolExecutor) -> Any:
|
||||
agent, task = pair
|
||||
return await run_agent_async(agent, task, executor)
|
||||
|
||||
cpu_cores = cpu_count()
|
||||
batch_size = batch_size or cpu_cores
|
||||
max_workers = max_workers or cpu_cores * 2
|
||||
results = []
|
||||
|
||||
try:
|
||||
loop = asyncio.get_event_loop()
|
||||
except RuntimeError:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
|
||||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||||
for i in range(0, len(agent_task_pairs), batch_size):
|
||||
batch = agent_task_pairs[i : i + batch_size]
|
||||
batch_results = loop.run_until_complete(
|
||||
asyncio.gather(*(run_pair_async(pair, executor) for pair in batch))
|
||||
)
|
||||
results.extend(batch_results)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
async def run_agent_with_timeout(
|
||||
agent: AgentType,
|
||||
task: str,
|
||||
timeout: float,
|
||||
executor: ThreadPoolExecutor
|
||||
) -> Any:
|
||||
"""
|
||||
Run an agent with a timeout limit.
|
||||
|
||||
Args:
|
||||
agent: Agent instance to run
|
||||
task: Task string to execute
|
||||
timeout: Timeout in seconds
|
||||
executor: ThreadPoolExecutor instance
|
||||
|
||||
Returns:
|
||||
Agent execution result or None if timeout occurs
|
||||
"""
|
||||
try:
|
||||
return await asyncio.wait_for(
|
||||
run_agent_async(agent, task, executor),
|
||||
timeout=timeout
|
||||
)
|
||||
except asyncio.TimeoutError:
|
||||
return None
|
||||
|
||||
@profile_func
|
||||
def run_agents_with_timeout(
|
||||
agents: List[AgentType],
|
||||
task: str,
|
||||
timeout: float,
|
||||
batch_size: int = None,
|
||||
max_workers: int = None
|
||||
) -> List[Any]:
|
||||
"""
|
||||
Run multiple agents concurrently with a timeout for each agent.
|
||||
|
||||
Args:
|
||||
agents: List of Agent instances
|
||||
task: Task string to execute
|
||||
timeout: Timeout in seconds for each agent
|
||||
batch_size: Number of agents to run in parallel
|
||||
max_workers: Maximum number of threads
|
||||
|
||||
Returns:
|
||||
List of outputs (None for timed out agents)
|
||||
"""
|
||||
cpu_cores = cpu_count()
|
||||
batch_size = batch_size or cpu_cores
|
||||
max_workers = max_workers or cpu_cores * 2
|
||||
results = []
|
||||
|
||||
try:
|
||||
loop = asyncio.get_event_loop()
|
||||
except RuntimeError:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
|
||||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||||
for i in range(0, len(agents), batch_size):
|
||||
batch = agents[i : i + batch_size]
|
||||
batch_results = loop.run_until_complete(
|
||||
asyncio.gather(
|
||||
*(run_agent_with_timeout(agent, task, timeout, executor)
|
||||
for agent in batch)
|
||||
)
|
||||
)
|
||||
results.extend(batch_results)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@dataclass
|
||||
class ResourceMetrics:
|
||||
cpu_percent: float
|
||||
memory_percent: float
|
||||
active_threads: int
|
||||
|
||||
def get_system_metrics() -> ResourceMetrics:
|
||||
"""Get current system resource usage"""
|
||||
return ResourceMetrics(
|
||||
cpu_percent=psutil.cpu_percent(),
|
||||
memory_percent=psutil.virtual_memory().percent,
|
||||
active_threads=threading.active_count()
|
||||
)
|
||||
|
||||
@profile_func
|
||||
def run_agents_with_resource_monitoring(
|
||||
agents: List[AgentType],
|
||||
task: str,
|
||||
cpu_threshold: float = 90.0,
|
||||
memory_threshold: float = 90.0,
|
||||
check_interval: float = 1.0
|
||||
) -> List[Any]:
|
||||
"""
|
||||
Run agents with system resource monitoring and adaptive batch sizing.
|
||||
|
||||
Args:
|
||||
agents: List of Agent instances
|
||||
task: Task string to execute
|
||||
cpu_threshold: Max CPU usage percentage
|
||||
memory_threshold: Max memory usage percentage
|
||||
check_interval: Resource check interval in seconds
|
||||
|
||||
Returns:
|
||||
List of outputs from each agent
|
||||
"""
|
||||
async def monitor_resources():
|
||||
while True:
|
||||
metrics = get_system_metrics()
|
||||
if metrics.cpu_percent > cpu_threshold or metrics.memory_percent > memory_threshold:
|
||||
# Reduce batch size or pause execution
|
||||
pass
|
||||
await asyncio.sleep(check_interval)
|
||||
|
||||
# Implementation details...
|
||||
|
||||
# # Example usage:
|
||||
# # Initialize your agents with the same model to avoid re-creating it
|
||||
# agents = [
|
||||
# Agent(
|
||||
# agent_name=f"Financial-Analysis-Agent_parallel_swarm{i}",
|
||||
# system_prompt=FINANCIAL_AGENT_SYS_PROMPT,
|
||||
# llm=model,
|
||||
# max_loops=1,
|
||||
# autosave=True,
|
||||
# dashboard=False,
|
||||
# verbose=False,
|
||||
# dynamic_temperature_enabled=False,
|
||||
# saved_state_path=f"finance_agent_{i}.json",
|
||||
# user_name="swarms_corp",
|
||||
# retry_attempts=1,
|
||||
# context_length=200000,
|
||||
# return_step_meta=False,
|
||||
# )
|
||||
# for i in range(5) # Assuming you want 10 agents
|
||||
# ]
|
||||
|
||||
# task = "How can I establish a ROTH IRA to buy stocks and get a tax break? What are the criteria"
|
||||
# outputs = run_agents_concurrently(agents, task)
|
||||
|
||||
# for i, output in enumerate(outputs):
|
||||
# print(f"Output from agent {i+1}:\n{output}")
|
||||
|
||||
@ -1,120 +0,0 @@
|
||||
import os
|
||||
|
||||
# from swarms.structs. import OpenAIChat
|
||||
import asyncio
|
||||
from swarms.utils.calculate_func_metrics import profile_func
|
||||
|
||||
|
||||
# Function to run a single agent on the task (synchronous)
|
||||
def run_single_agent(agent, task):
|
||||
return agent.run(task)
|
||||
|
||||
|
||||
# Asynchronous wrapper for agent tasks
|
||||
async def run_agent_async(agent, task):
|
||||
loop = asyncio.get_event_loop()
|
||||
return await loop.run_in_executor(
|
||||
None, run_single_agent, agent, task
|
||||
)
|
||||
|
||||
|
||||
# Asynchronous function to run agents concurrently
|
||||
async def run_agents_concurrently_async(agents, task: str):
|
||||
"""
|
||||
Run multiple agents concurrently on the same task with optimized performance.
|
||||
|
||||
:param agents: List of Agent instances to run concurrently.
|
||||
:param task: The task string to execute by all agents.
|
||||
:return: A list of outputs from each agent.
|
||||
"""
|
||||
|
||||
# Run all agents asynchronously using asyncio.gather
|
||||
results = await asyncio.gather(
|
||||
*(run_agent_async(agent, task) for agent in agents)
|
||||
)
|
||||
return results
|
||||
|
||||
|
||||
# Function to manage the overall process and batching
|
||||
@profile_func
|
||||
def run_agents_concurrently(agents, task: str, batch_size: int = 5):
|
||||
"""
|
||||
Manage and run multiple agents concurrently in batches, with optimized performance.
|
||||
|
||||
:param agents: List of Agent instances to run concurrently.
|
||||
:param task: The task string to execute by all agents.
|
||||
:param batch_size: Number of agents to run in parallel in each batch.
|
||||
:return: A list of outputs from each agent.
|
||||
"""
|
||||
|
||||
results = []
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
batch_size = (
|
||||
os.cpu_count() if batch_size > os.cpu_count() else batch_size
|
||||
)
|
||||
|
||||
# Process agents in batches to avoid overwhelming system resources
|
||||
for i in range(0, len(agents), batch_size):
|
||||
batch = agents[i : i + batch_size]
|
||||
batch_results = loop.run_until_complete(
|
||||
run_agents_concurrently_async(batch, task)
|
||||
)
|
||||
results.extend(batch_results)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
# # Example usage:
|
||||
# # Initialize your agents with the same model to avoid re-creating it
|
||||
# agents = [
|
||||
# Agent(
|
||||
# agent_name=f"Financial-Analysis-Agent_parallel_swarm{i}",
|
||||
# system_prompt=FINANCIAL_AGENT_SYS_PROMPT,
|
||||
# llm=model,
|
||||
# max_loops=1,
|
||||
# autosave=True,
|
||||
# dashboard=False,
|
||||
# verbose=False,
|
||||
# dynamic_temperature_enabled=False,
|
||||
# saved_state_path=f"finance_agent_{i}.json",
|
||||
# user_name="swarms_corp",
|
||||
# retry_attempts=1,
|
||||
# context_length=200000,
|
||||
# return_step_meta=False,
|
||||
# )
|
||||
# for i in range(5) # Assuming you want 10 agents
|
||||
# ]
|
||||
|
||||
# task = "How can I establish a ROTH IRA to buy stocks and get a tax break? What are the criteria"
|
||||
# outputs = run_agents_concurrently(agents, task)
|
||||
|
||||
# for i, output in enumerate(outputs):
|
||||
# print(f"Output from agent {i+1}:\n{output}")
|
||||
|
||||
# # Output from agent 2:
|
||||
# # execution_time=12.89196228981018 memory_usage=-294.9375 cpu_usage=-10.3 io_operations=23309 function_calls=1
|
||||
# # execution_time=11.810921907424927 memory_usage=-242.734375 cpu_usage=-26.4 io_operations=10752 function_calls=1
|
||||
|
||||
# # Parallel
|
||||
# # execution_time=18.79391312599182 memory_usage=-342.9375 cpu_usage=-2.5 io_operations=59518 function_calls=1
|
||||
|
||||
# # # Multiprocess
|
||||
# # 2024-08-22T14:49:33.986491-0400 Function metrics: {
|
||||
# # "execution_time": 24.783875942230225,
|
||||
# # "memory_usage": -286.734375,
|
||||
# # "cpu_usage": -24.6,
|
||||
# # "io_operations": 17961,
|
||||
# # "function_calls": 1
|
||||
# # }
|
||||
|
||||
|
||||
# # Latest
|
||||
# # Analysis-Agent_parallel_swarm4_state.json
|
||||
# # 2024-08-22T15:43:11.800970-0400 Function metrics: {
|
||||
# # "execution_time": 11.062030792236328,
|
||||
# # "memory_usage": -249.5625,
|
||||
# # "cpu_usage": -15.700000000000003,
|
||||
# # "io_operations": 13439,
|
||||
# # "function_calls": 1
|
||||
# # }
|
||||
@ -1,130 +0,0 @@
|
||||
import asyncio
|
||||
from typing import List, Any
|
||||
from swarms import Agent
|
||||
from multiprocessing import cpu_count
|
||||
from swarms.utils.calculate_func_metrics import profile_func
|
||||
|
||||
# Use uvloop for faster asyncio event loop
|
||||
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
|
||||
|
||||
|
||||
def run_single_agent(agent: Agent, task: str) -> Any:
|
||||
"""
|
||||
Run a single agent on the given task.
|
||||
|
||||
Args:
|
||||
agent (Agent): The agent to run.
|
||||
task (str): The task for the agent to perform.
|
||||
|
||||
Returns:
|
||||
Any: The result of the agent's execution.
|
||||
"""
|
||||
return agent.run(task)
|
||||
|
||||
|
||||
async def run_agent_async(agent: Agent, task: str) -> Any:
|
||||
"""
|
||||
Asynchronous wrapper for agent tasks.
|
||||
|
||||
Args:
|
||||
agent (Agent): The agent to run asynchronously.
|
||||
task (str): The task for the agent to perform.
|
||||
|
||||
Returns:
|
||||
Any: The result of the agent's execution.
|
||||
"""
|
||||
loop = asyncio.get_event_loop()
|
||||
return await loop.run_in_executor(
|
||||
None, run_single_agent, agent, task
|
||||
)
|
||||
|
||||
|
||||
async def run_agents_concurrently_async(
|
||||
agents: List[Agent], task: str
|
||||
) -> List[Any]:
|
||||
"""
|
||||
Run multiple agents concurrently on the same task with optimized performance.
|
||||
|
||||
Args:
|
||||
agents (List[Agent]): List of Agent instances to run concurrently.
|
||||
task (str): The task string to execute by all agents.
|
||||
|
||||
Returns:
|
||||
List[Any]: A list of outputs from each agent.
|
||||
"""
|
||||
results = await asyncio.gather(
|
||||
*(run_agent_async(agent, task) for agent in agents)
|
||||
)
|
||||
return results
|
||||
|
||||
|
||||
@profile_func
|
||||
def run_agents_concurrently_multiprocess(
|
||||
agents: List[Agent], task: str, batch_size: int = cpu_count()
|
||||
) -> List[Any]:
|
||||
"""
|
||||
Manage and run multiple agents concurrently in batches, with optimized performance.
|
||||
|
||||
Args:
|
||||
agents (List[Agent]): List of Agent instances to run concurrently.
|
||||
task (str): The task string to execute by all agents.
|
||||
batch_size (int, optional): Number of agents to run in parallel in each batch.
|
||||
Defaults to the number of CPU cores.
|
||||
|
||||
Returns:
|
||||
List[Any]: A list of outputs from each agent.
|
||||
"""
|
||||
results = []
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
# Process agents in batches to avoid overwhelming system resources
|
||||
for i in range(0, len(agents), batch_size):
|
||||
batch = agents[i : i + batch_size]
|
||||
batch_results = loop.run_until_complete(
|
||||
run_agents_concurrently_async(batch, task)
|
||||
)
|
||||
results.extend(batch_results)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
# # # Example usage:
|
||||
# # Initialize your agents with the same model to avoid re-creating it
|
||||
# agents = [
|
||||
# Agent(
|
||||
# agent_name=f"Financial-Analysis-Agent_new_parallel_swarm_test{i}",
|
||||
# system_prompt=FINANCIAL_AGENT_SYS_PROMPT,
|
||||
# llm=model,
|
||||
# max_loops=1,
|
||||
# autosave=True,
|
||||
# dashboard=False,
|
||||
# verbose=False,
|
||||
# dynamic_temperature_enabled=False,
|
||||
# saved_state_path=f"finance_agent_{i}.json",
|
||||
# user_name="swarms_corp",
|
||||
# retry_attempts=1,
|
||||
# context_length=200000,
|
||||
# return_step_meta=False,
|
||||
# )
|
||||
# for i in range(5) # Assuming you want 10 agents
|
||||
# ]
|
||||
|
||||
# task = "How can I establish a ROTH IRA to buy stocks and get a tax break? What are the criteria"
|
||||
# outputs = run_agents_concurrently_multiprocess(
|
||||
# agents,
|
||||
# task,
|
||||
# )
|
||||
|
||||
# for i, output in enumerate(outputs):
|
||||
# print(f"Output from agent {i+1}:\n{output}")
|
||||
|
||||
|
||||
# # execution_time=15.958749055862427 memory_usage=-328.046875 cpu_usage=-2.5999999999999943 io_operations=81297 function_calls=1
|
||||
# # Analysis-Agent_new_parallel_swarm_test1_state.json
|
||||
# # 2024-08-22T15:42:12.463246-0400 Function metrics: {
|
||||
# # "execution_time": 15.958749055862427,
|
||||
# # "memory_usage": -328.046875,
|
||||
# # "cpu_usage": -2.5999999999999943,
|
||||
# # "io_operations": 81297,
|
||||
# "function_calls": 1
|
||||
# }
|
||||
Loading…
Reference in new issue