[EXAMPLES][8.4.0 New Examples]

pull/1100/head
Kye Gomez 1 week ago
parent 5ec8fdda0a
commit daf261cc73

@ -0,0 +1,73 @@
from swarms import Agent, AgentRearrange
# Create specialized quantitative research agents
weather_data_agent = Agent(
agent_name="Weather-Data-Agent",
agent_description="Expert in weather data collection, agricultural commodity research, and meteorological analysis",
model_name="claude-sonnet-4-20250514",
max_loops=1,
system_prompt="""You are a quantitative weather data research specialist. Your role is to:
1. Collect and analyze weather data from multiple sources (NOAA, Weather APIs, satellite data)
2. Research agricultural commodity markets and their weather dependencies
3. Identify weather patterns that historically impact crop yields and commodity prices
4. Gather data on seasonal weather trends, precipitation patterns, temperature anomalies
5. Research specific regions and their agricultural production cycles
6. Collect data on extreme weather events and their market impact
7. Analyze historical correlations between weather data and commodity price movements
Focus on actionable weather intelligence for trading opportunities. Always provide specific data points,
timeframes, and geographic regions. Include confidence levels and data quality assessments.""",
)
quant_analysis_agent = Agent(
agent_name="Quant-Analysis-Agent",
agent_description="Expert in quantitative analysis of weather patterns, arbitrage opportunities, and statistical modeling",
model_name="claude-sonnet-4-20250514",
max_loops=1,
system_prompt="""You are a quantitative analysis specialist focused on weather-driven arbitrage opportunities. Your role is to:
1. Analyze weather data correlations with commodity price movements
2. Identify statistical arbitrage opportunities in agricultural futures markets
3. Calculate risk-adjusted returns for weather-based trading strategies
4. Model price impact scenarios based on weather forecasts
5. Identify seasonal patterns and mean reversion opportunities
6. Analyze basis risk and correlation breakdowns between weather and prices
7. Calculate optimal position sizes and hedging ratios
8. Assess market inefficiencies in weather-sensitive commodities
Focus on actionable trading signals with specific entry/exit criteria, risk metrics, and expected returns.
Always provide quantitative justification and statistical confidence levels.""",
)
trading_strategy_agent = Agent(
agent_name="Trading-Strategy-Agent",
agent_description="Expert in trading strategy development, risk assessment, and portfolio management for weather-driven arbitrage",
model_name="claude-sonnet-4-20250514",
max_loops=1,
system_prompt="""You are a quantitative trading strategy specialist focused on weather-driven arbitrage opportunities. Your role is to:
1. Develop comprehensive trading strategies based on weather data and commodity analysis
2. Create detailed risk management frameworks for weather-sensitive positions
3. Design portfolio allocation strategies for agricultural commodity arbitrage
4. Develop hedging strategies to mitigate weather-related risks
5. Create position sizing models based on volatility and correlation analysis
6. Design entry and exit criteria for weather-based trades
7. Develop contingency plans for unexpected weather events
8. Create performance monitoring and evaluation frameworks
Focus on practical, implementable trading strategies with clear risk parameters,
position management rules, and performance metrics. Always include specific trade setups,
risk limits, and monitoring protocols.""",
)
rearrange_system = AgentRearrange(
agents=[
weather_data_agent,
quant_analysis_agent,
trading_strategy_agent,
],
flow=f"{trading_strategy_agent.agent_name} -> {quant_analysis_agent.agent_name}, {weather_data_agent.agent_name}",
max_loops=1,
)
rearrange_system.run(
"What are the best weather trades for the rest of the year 2025? Can we short wheat futures, corn futures, soybean futures, etc.?"
)

@ -0,0 +1,17 @@
import json
from swarms import AutoSwarmBuilder
swarm = AutoSwarmBuilder(
name="My Swarm",
description="A swarm of agents",
verbose=True,
max_loops=1,
model_name="gpt-4o-mini",
execution_type="return-agents",
)
out = swarm.run(
task="Create an accounting team to analyze crypto transactions, there must be 5 agents in the team with extremely extensive prompts. Make the prompts extremely detailed and specific and long and comprehensive. Make sure to include all the details of the task in the prompts."
)
print(json.dumps(out, indent=4))

@ -0,0 +1,19 @@
from swarms import Agent
# Initialize the agent
agent = Agent(
agent_name="Quantitative-Trading-Agent",
agent_description="Advanced quantitative trading and algorithmic analysis agent",
model_name="claude-sonnet-4-2025051eqfewfwmfkmekef",
dynamic_temperature_enabled=True,
max_loops=1,
dynamic_context_window=True,
streaming_on=True,
fallback_models=["gpt-4o-mini", "anthropic/claude-sonnet-4-5"],
)
out = agent.run(
task="What are the top five best energy stocks across nuclear, solar, gas, and other energy sources?",
)
print(out)

@ -0,0 +1,106 @@
from swarms import Agent
from swarms.structs.aop import (
AOP,
)
# Create specialized agents
research_agent = Agent(
agent_name="Research-Agent",
agent_description="Expert in research, data collection, and information gathering",
model_name="anthropic/claude-sonnet-4-5",
max_loops=1,
top_p=None,
dynamic_temperature_enabled=True,
system_prompt="""You are a research specialist. Your role is to:
1. Gather comprehensive information on any given topic
2. Analyze data from multiple sources
3. Provide well-structured research findings
4. Cite sources and maintain accuracy
5. Present findings in a clear, organized manner
Always provide detailed, factual information with proper context.""",
)
analysis_agent = Agent(
agent_name="Analysis-Agent",
agent_description="Expert in data analysis, pattern recognition, and generating insights",
model_name="anthropic/claude-sonnet-4-5",
max_loops=1,
top_p=None,
dynamic_temperature_enabled=True,
system_prompt="""You are an analysis specialist. Your role is to:
1. Analyze data and identify patterns
2. Generate actionable insights
3. Create visualizations and summaries
4. Provide statistical analysis
5. Make data-driven recommendations
Focus on extracting meaningful insights from information.""",
)
writing_agent = Agent(
agent_name="Writing-Agent",
agent_description="Expert in content creation, editing, and communication",
model_name="anthropic/claude-sonnet-4-5",
max_loops=1,
top_p=None,
dynamic_temperature_enabled=True,
system_prompt="""You are a writing specialist. Your role is to:
1. Create engaging, well-structured content
2. Edit and improve existing text
3. Adapt tone and style for different audiences
4. Ensure clarity and coherence
5. Follow best practices in writing
Always produce high-quality, professional content.""",
)
code_agent = Agent(
agent_name="Code-Agent",
agent_description="Expert in programming, code review, and software development",
model_name="anthropic/claude-sonnet-4-5",
max_loops=1,
top_p=None,
dynamic_temperature_enabled=True,
system_prompt="""You are a coding specialist. Your role is to:
1. Write clean, efficient code
2. Debug and fix issues
3. Review and optimize code
4. Explain programming concepts
5. Follow best practices and standards
Always provide working, well-documented code.""",
)
financial_agent = Agent(
agent_name="Financial-Agent",
agent_description="Expert in financial analysis, market research, and investment insights",
model_name="anthropic/claude-sonnet-4-5",
max_loops=1,
top_p=None,
dynamic_temperature_enabled=True,
system_prompt="""You are a financial specialist. Your role is to:
1. Analyze financial data and markets
2. Provide investment insights
3. Assess risk and opportunities
4. Create financial reports
5. Explain complex financial concepts
Always provide accurate, well-reasoned financial analysis.""",
)
# Basic usage - individual agent addition
deployer = AOP("MyAgentServer", verbose=True, port=5932)
agents = [
research_agent,
analysis_agent,
writing_agent,
code_agent,
financial_agent,
]
deployer.add_agents_batch(agents)
deployer.run()

@ -1,6 +1,6 @@
import json import json
import traceback import traceback
from typing import Any, List, Optional, Tuple from typing import Any, List, Optional
from dotenv import load_dotenv from dotenv import load_dotenv
from loguru import logger from loguru import logger
@ -14,6 +14,13 @@ from swarms.utils.litellm_wrapper import LiteLLM
load_dotenv() load_dotenv()
execution_types = [
"return-agents",
"execute-swarm-router",
"return-swarm-router-config",
"return-agents-objects",
]
BOSS_SYSTEM_PROMPT = """ BOSS_SYSTEM_PROMPT = """
You are an expert multi-agent architecture designer and team coordinator. Your role is to create and orchestrate sophisticated teams of specialized AI agents, each with distinct personalities, roles, and capabilities. Your primary goal is to ensure the multi-agent system operates efficiently while maintaining clear communication, well-defined responsibilities, and optimal task distribution. You are an expert multi-agent architecture designer and team coordinator. Your role is to create and orchestrate sophisticated teams of specialized AI agents, each with distinct personalities, roles, and capabilities. Your primary goal is to ensure the multi-agent system operates efficiently while maintaining clear communication, well-defined responsibilities, and optimal task distribution.
@ -143,23 +150,32 @@ class AgentSpec(BaseModel):
description="The initial instruction or context provided to the agent, guiding its behavior and responses during execution.", description="The initial instruction or context provided to the agent, guiding its behavior and responses during execution.",
) )
model_name: Optional[str] = Field( model_name: Optional[str] = Field(
description="The name of the AI model that the agent will utilize for processing tasks and generating outputs. For example: gpt-4o, gpt-4o-mini, openai/o3-mini" "gpt-4.1",
description="The name of the AI model that the agent will utilize for processing tasks and generating outputs. For example: gpt-4o, gpt-4o-mini, openai/o3-mini",
) )
auto_generate_prompt: Optional[bool] = Field( auto_generate_prompt: Optional[bool] = Field(
description="A flag indicating whether the agent should automatically create prompts based on the task requirements." False,
description="A flag indicating whether the agent should automatically create prompts based on the task requirements.",
) )
max_tokens: Optional[int] = Field( max_tokens: Optional[int] = Field(
None, 8192,
description="The maximum number of tokens that the agent is allowed to generate in its responses, limiting output length.", description="The maximum number of tokens that the agent is allowed to generate in its responses, limiting output length.",
) )
temperature: Optional[float] = Field( temperature: Optional[float] = Field(
description="A parameter that controls the randomness of the agent's output; lower values result in more deterministic responses." 0.5,
description="A parameter that controls the randomness of the agent's output; lower values result in more deterministic responses.",
) )
role: Optional[str] = Field( role: Optional[str] = Field(
description="The designated role of the agent within the swarm, which influences its behavior and interaction with other agents." "worker",
description="The designated role of the agent within the swarm, which influences its behavior and interaction with other agents.",
) )
max_loops: Optional[int] = Field( max_loops: Optional[int] = Field(
description="The maximum number of times the agent is allowed to repeat its task, enabling iterative processing if necessary." 1,
description="The maximum number of times the agent is allowed to repeat its task, enabling iterative processing if necessary.",
)
goal: Optional[str] = Field(
None,
description="The primary objective or desired outcome the agent is tasked with achieving.",
) )
@ -171,57 +187,10 @@ class Agents(BaseModel):
) )
execution_types = [
"return-agents",
"execute-swarm-router",
"return-swarm-router-config",
"return-agent-configurations",
"return-agent-specs",
"return-agent-dictionary",
]
class AgentConfig(BaseModel):
"""Configuration for an individual agent in a swarm"""
name: str = Field(
description="The name of the agent. This should be a unique identifier that distinguishes this agent from others within the swarm. The name should reflect the agent's primary function, role, or area of expertise, and should be easily recognizable by both humans and other agents in the system. A well-chosen name helps clarify the agent's responsibilities and facilitates effective communication and collaboration within the swarm.",
)
description: str = Field(
description=(
"A comprehensive description of the agent's purpose, core responsibilities, and capabilities within the swarm. One sentence is enough."
),
)
system_prompt: str = Field(
description=(
"The system prompt that defines the agent's behavior. This prompt should be extremely long, comprehensive, and extensive, encapsulating the agent's identity, operational guidelines, and decision-making framework in great detail. It provides the foundational instructions that guide the agent's actions, communication style, and interaction protocols with both users and other agents. The system prompt should be highly detailed, unambiguous, and exhaustive, ensuring the agent consistently acts in accordance with its intended role and adheres to the swarm's standards and best practices. The prompt should leave no ambiguity and cover all relevant aspects of the agent's responsibilities, behaviors, and expected outcomes."
),
)
goal: str = Field(
description="The goal of the agent. This should clearly state the primary objective or desired outcome the agent is tasked with achieving. The goal should be specific, measurable, and aligned with the overall mission of the swarm. It serves as the guiding principle for the agent's actions and decision-making processes, helping to maintain focus and drive effective collaboration within the multi-agent system.",
)
model_name: str = Field(
description="The model to use for the agent. This is the model that will be used to generate the agent's responses. For example, 'gpt-4o-mini' or 'claude-sonnet-3.7-sonnet-20240620'."
)
temperature: float = Field(
description="The temperature to use for the agent. This controls the randomness of the agent's responses. For example, 0.5 or 1.0."
)
max_loops: int = Field(
description="The maximum number of loops for the agent to run. This is the maximum number of times the agent will run its loop. For example, 1, 2, or 3. Keep this set to 1 unless the agent requires more than one loop to complete its task.",
)
# max_loops: int = Field(
# description="The maximum number of loops for the agent to run",
# )
class Config:
arbitrary_types_allowed = True
class AgentsConfig(BaseModel): class AgentsConfig(BaseModel):
"""Configuration for a list of agents in a swarm""" """Configuration for a list of agents in a swarm"""
agents: List[AgentConfig] = Field( agents: List[AgentSpec] = Field(
description="A list of agent configurations", description="A list of agent configurations",
) )
@ -233,7 +202,7 @@ class SwarmRouterConfig(BaseModel):
description: str = Field( description: str = Field(
description="Description of the team of agents" description="Description of the team of agents"
) )
agents: List[AgentConfig] = Field( agents: List[AgentSpec] = Field(
description="A list of agent configurations", description="A list of agent configurations",
) )
swarm_type: SwarmType = Field( swarm_type: SwarmType = Field(
@ -245,7 +214,10 @@ class SwarmRouterConfig(BaseModel):
rules: Optional[str] = Field( rules: Optional[str] = Field(
description="Rules to inject into every agent. This is a string of rules that will be injected into every agent's system prompt. This is a good place to put things like 'You are a helpful assistant' or 'You are a helpful assistant that can answer questions and help with tasks'." description="Rules to inject into every agent. This is a string of rules that will be injected into every agent's system prompt. This is a good place to put things like 'You are a helpful assistant' or 'You are a helpful assistant that can answer questions and help with tasks'."
) )
multi_agent_collab_prompt: Optional[str] = Field(
None,
description="Prompt for multi-agent collaboration and coordination.",
)
task: str = Field( task: str = Field(
description="The task to be executed by the swarm", description="The task to be executed by the swarm",
) )
@ -271,7 +243,6 @@ class AutoSwarmBuilder:
interactive (bool): Whether to enable interactive mode. Defaults to False. interactive (bool): Whether to enable interactive mode. Defaults to False.
max_tokens (int): Maximum tokens for the LLM responses. Defaults to 8000. max_tokens (int): Maximum tokens for the LLM responses. Defaults to 8000.
execution_type (str): Type of execution to perform. Defaults to "return-agents". execution_type (str): Type of execution to perform. Defaults to "return-agents".
return_dictionary (bool): Whether to return dictionary format for agent specs. Defaults to True.
system_prompt (str): System prompt for the boss agent. Defaults to BOSS_SYSTEM_PROMPT. system_prompt (str): System prompt for the boss agent. Defaults to BOSS_SYSTEM_PROMPT.
""" """
@ -286,8 +257,8 @@ class AutoSwarmBuilder:
interactive: bool = False, interactive: bool = False,
max_tokens: int = 8000, max_tokens: int = 8000,
execution_type: execution_types = "return-agents", execution_type: execution_types = "return-agents",
return_dictionary: bool = True,
system_prompt: str = BOSS_SYSTEM_PROMPT, system_prompt: str = BOSS_SYSTEM_PROMPT,
additional_llm_args: dict = {},
): ):
"""Initialize the AutoSwarmBuilder. """Initialize the AutoSwarmBuilder.
@ -313,15 +284,19 @@ class AutoSwarmBuilder:
self.interactive = interactive self.interactive = interactive
self.max_tokens = max_tokens self.max_tokens = max_tokens
self.execution_type = execution_type self.execution_type = execution_type
self.return_dictionary = return_dictionary
self.system_prompt = system_prompt self.system_prompt = system_prompt
self.additional_llm_args = additional_llm_args
self.conversation = Conversation() self.conversation = Conversation()
self.agents_pool = [] self.agents_pool = []
self.reliability_check() self.reliability_check()
def reliability_check(self): def reliability_check(self):
"""Perform reliability checks on the AutoSwarmBuilder configuration.
Raises:
ValueError: If max_loops is set to 0
"""
if self.max_loops == 0: if self.max_loops == 0:
raise ValueError( raise ValueError(
f"AutoSwarmBuilder: {self.name} max_loops cannot be 0" f"AutoSwarmBuilder: {self.name} max_loops cannot be 0"
@ -332,9 +307,20 @@ class AutoSwarmBuilder:
) )
def _execute_task(self, task: str): def _execute_task(self, task: str):
"""Execute a task by creating agents and initializing the swarm router.
Args:
task (str): The task to execute
Returns:
Any: The result of the swarm router execution
"""
logger.info(f"Executing task: {task}") logger.info(f"Executing task: {task}")
agents = self.create_agents(task) agents_dict = self.create_agents(task)
# Convert dictionary to Agent objects for execution
agents = self.create_agents_from_specs(agents_dict)
if self.execution_type == "return-agents": if self.execution_type == "return-agents":
logger.info("Setting random models for agents") logger.info("Setting random models for agents")
@ -342,50 +328,20 @@ class AutoSwarmBuilder:
return self.initialize_swarm_router(agents=agents, task=task) return self.initialize_swarm_router(agents=agents, task=task)
def run(self, task: str, *args, **kwargs): def dict_to_agent(self, output: dict):
"""Run the swarm on a given task. """Convert dictionary output to Agent objects.
Args: Args:
task (str): The task to execute output (dict): Dictionary containing agent configurations
*args: Additional positional arguments
**kwargs: Additional keyword arguments
Returns: Returns:
Any: The result of the swarm execution List[Agent]: List of created Agent objects
Raises:
Exception: If there's an error during execution
""" """
try:
if self.execution_type == "return-swarm-router-config":
return self.create_router_config(task)
elif self.execution_type == "return-agent-configurations":
return self.create_agents(task)
elif self.execution_type == "return-agent-specs":
return self._create_agent_specs(task)
elif self.execution_type == "return-agent-dictionary":
return self._create_agent_dictionary(task)
else:
return self._execute_task(task)
except Exception as e:
logger.error(
f"AutoSwarmBuilder: Error in swarm execution: {str(e)} Traceback: {traceback.format_exc()}",
exc_info=True,
)
raise
def dict_to_agent(self, output: dict):
agents = [] agents = []
if isinstance(output, dict): if isinstance(output, dict):
for agent_config in output["agents"]: for agent_config in output["agents"]:
logger.info(f"Building agent: {agent_config['name']}") logger.info(f"Building agent: {agent_config['name']}")
agent = self.build_agent( agent = Agent(**agent_config)
agent_name=agent_config["name"],
agent_description=agent_config["description"],
agent_system_prompt=agent_config["system_prompt"],
)
agents.append(agent) agents.append(agent)
logger.info( logger.info(
f"Successfully built agent: {agent_config['name']}" f"Successfully built agent: {agent_config['name']}"
@ -417,12 +373,21 @@ class AutoSwarmBuilder:
raise e raise e
def build_llm_agent(self, config: BaseModel): def build_llm_agent(self, config: BaseModel):
"""Build a LiteLLM agent with the specified configuration.
Args:
config (BaseModel): Pydantic model configuration for the LLM
Returns:
LiteLLM: Configured LiteLLM instance
"""
return LiteLLM( return LiteLLM(
model_name=self.model_name, model_name=self.model_name,
system_prompt=BOSS_SYSTEM_PROMPT, system_prompt=BOSS_SYSTEM_PROMPT,
temperature=0.5, temperature=0.5,
response_format=config, response_format=config,
max_tokens=self.max_tokens, max_tokens=self.max_tokens,
**self.additional_llm_args,
) )
def create_agents(self, task: str): def create_agents(self, task: str):
@ -432,21 +397,18 @@ class AutoSwarmBuilder:
task (str): The task to create agents for task (str): The task to create agents for
Returns: Returns:
List[Agent]: List of created agents dict: Dictionary containing agent specifications
Raises: Raises:
Exception: If there's an error during agent creation Exception: If there's an error during agent creation
""" """
try: try:
model = self.build_llm_agent(config=AgentsConfig) logger.info("Creating agents from specifications")
model = self.build_llm_agent(config=Agents)
output = model.run(
f"Create the agents for the following task: {task}"
)
output = json.loads(output) agents_dictionary = model.run(task)
return output return agents_dictionary
except Exception as e: except Exception as e:
logger.error( logger.error(
@ -455,43 +417,6 @@ class AutoSwarmBuilder:
) )
raise e raise e
def build_agent(
self,
agent_name: str,
agent_description: str,
agent_system_prompt: str,
) -> Agent:
"""Build a single agent with enhanced error handling.
Args:
agent_name (str): Name of the agent
agent_description (str): Description of the agent
agent_system_prompt (str): System prompt for the agent
Returns:
Agent: The constructed agent
Raises:
Exception: If there's an error during agent construction
"""
logger.info(f"Building agent: {agent_name}")
try:
agent = Agent(
agent_name=agent_name,
agent_description=agent_description,
system_prompt=agent_system_prompt,
verbose=self.verbose,
dynamic_temperature_enabled=False,
)
logger.info(f"Successfully built agent: {agent_name}")
return agent
except Exception as e:
logger.error(
f"Error building agent {agent_name}: {str(e)}",
exc_info=True,
)
raise
def initialize_swarm_router(self, agents: List[Agent], task: str): def initialize_swarm_router(self, agents: List[Agent], task: str):
"""Initialize and run the swarm router. """Initialize and run the swarm router.
@ -552,75 +477,8 @@ class AutoSwarmBuilder:
Raises: Raises:
Exception: If there's an error during batch execution Exception: If there's an error during batch execution
""" """
return [self.run(task) for task in tasks] return [self.run(task) for task in tasks]
def _create_agent_specs(
self, task: str
) -> Tuple[List[Agent], int]:
"""Create agent specifications for a given task.
Args:
task (str): The task to create agents for
Returns:
Tuple[List[Agent], int]: List of created agents and count
"""
logger.info("Creating agent specifications for task")
agents = self._create_agents_from_specs(task)
return agents, len(agents)
def _create_agent_dictionary(self, task: str):
"""Create agent dictionary for a given task.
Args:
task (str): The task to create agents for
Returns:
dict: Dictionary containing agent configurations
"""
logger.info("Creating agent dictionary for task")
agents_dictionary = self._create_agents_from_specs(
task, return_dict=True
)
return agents_dictionary
def _create_agents_from_specs(
self, task: str, return_dict: bool = False
):
"""Create agents from specifications.
Args:
task (str): The task to create agents for
return_dict (bool): Whether to return dictionary format
Returns:
List[Agent] or dict: Created agents or dictionary
"""
logger.info("Creating agents from specifications")
model = LiteLLM(
model_name=self.model_name,
system_prompt=self.system_prompt,
temperature=0.1,
response_format=Agents,
max_tokens=8192,
)
agents_dictionary = model.run(task)
print(agents_dictionary)
print(type(agents_dictionary))
logger.info("Agents successfully created")
logger.info(f"Agents: {len(agents_dictionary.agents)}")
if return_dict or self.return_dictionary:
logger.info("Returning dictionary")
# Convert swarm config to dictionary
agents_dictionary = agents_dictionary.model_dump()
return agents_dictionary
else:
logger.info("Returning agents")
return self.create_agents_from_specs(agents_dictionary)
def create_agents_from_specs( def create_agents_from_specs(
self, agents_dictionary: Any self, agents_dictionary: Any
) -> List[Agent]: ) -> List[Agent]:
@ -634,75 +492,62 @@ class AutoSwarmBuilder:
""" """
# Create agents from config # Create agents from config
agents = [] agents = []
for agent_config in agents_dictionary.agents:
# Handle both dict and object formats
if isinstance(agents_dictionary, dict):
agents_list = agents_dictionary.get("agents", [])
else:
agents_list = agents_dictionary.agents
for agent_config in agents_list:
# Convert dict to AgentSpec if needed # Convert dict to AgentSpec if needed
if isinstance(agent_config, dict): if isinstance(agent_config, dict):
agent_config = AgentSpec(**agent_config) agent_config = AgentSpec(**agent_config)
agent = self.build_agent_from_spec( agent = Agent(**agent_config)
agent_name=agent_config.agent_name,
agent_description=agent_config.description,
agent_system_prompt=agent_config.system_prompt,
model_name=agent_config.model_name,
max_loops=agent_config.max_loops,
dynamic_temperature_enabled=True,
auto_generate_prompt=agent_config.auto_generate_prompt,
role=agent_config.role,
max_tokens=agent_config.max_tokens,
temperature=agent_config.temperature,
)
agents.append(agent) agents.append(agent)
return agents return agents
def build_agent_from_spec( def list_types(self):
self, """List all available execution types.
agent_name: str,
agent_description: str, Returns:
agent_system_prompt: str, List[str]: List of available execution types
max_loops: int = 1, """
model_name: str = "gpt-4.1", return execution_types
dynamic_temperature_enabled: bool = True,
auto_generate_prompt: bool = False, def run(self, task: str, *args, **kwargs):
role: str = "worker", """Run the swarm on a given task.
max_tokens: int = 8192,
temperature: float = 0.5,
) -> Agent:
"""Build a single agent from agent specification.
Args: Args:
agent_name (str): Name of the agent task (str): The task to execute
agent_description (str): Description of the agent's purpose *args: Additional positional arguments
agent_system_prompt (str): The system prompt for the agent **kwargs: Additional keyword arguments
max_loops (int): Maximum number of loops
model_name (str): Model name to use
dynamic_temperature_enabled (bool): Whether to enable dynamic temperature
auto_generate_prompt (bool): Whether to auto-generate prompts
role (str): Role of the agent
max_tokens (int): Maximum tokens
temperature (float): Temperature setting
Returns: Returns:
Agent: The constructed agent instance Any: The result of the swarm execution
Raises:
Exception: If there's an error during execution
""" """
logger.info(f"Building agent from spec: {agent_name}") try:
agent = Agent(
agent_name=agent_name,
description=agent_description,
system_prompt=agent_system_prompt,
model_name=model_name,
max_loops=max_loops,
dynamic_temperature_enabled=dynamic_temperature_enabled,
context_length=200000,
output_type="str",
streaming_on=False,
auto_generate_prompt=auto_generate_prompt,
role=role,
max_tokens=max_tokens,
temperature=temperature,
)
return agent
def list_types(self): if self.execution_type == "return-agents":
return execution_types return self.create_agents(task)
elif self.execution_type == "return-swarm-router-config":
return self.create_router_config(task)
elif self.execution_type == "return-agents-objects":
agents = self.create_agents(task)
return self.create_agents_from_specs(agents)
else:
raise ValueError(
f"Invalid execution type: {self.execution_type}"
)
except Exception as e:
logger.error(
f"AutoSwarmBuilder: Error in swarm execution: {str(e)} Traceback: {traceback.format_exc()}",
exc_info=True,
)
raise e

Loading…
Cancel
Save