[REMOVED][AgentsBuilder][MIGRATION][AgentsBuilder -> AutoSwarmBuilder]

pull/966/merge
Kye Gomez 1 week ago
parent 4650c5d70e
commit af9bc7e30b

@ -290,7 +290,6 @@ nav:
- Overview: "swarms/structs/multi_swarm_orchestration.md"
- HierarchicalSwarm: "swarms/structs/hierarchical_swarm.md"
- Hierarchical Structured Communication Framework: "swarms/structs/hierarchical_structured_communication_framework.md"
- Auto Agent Builder: "swarms/structs/auto_agent_builder.md"
- Hybrid Hierarchical-Cluster Swarm: "swarms/structs/hhcs.md"
- Auto Swarm Builder: "swarms/structs/auto_swarm_builder.md"
- Swarm Matcher: "swarms/structs/swarm_matcher.md"

@ -1,200 +0,0 @@
# Agent Builder
The Agent Builder is a powerful class that automatically builds and manages swarms of AI agents. It provides a flexible and extensible framework for creating, coordinating, and executing multiple AI agents working together to accomplish complex tasks.
## Overview
The Agent Builder uses a boss agent to delegate work and create new specialized agents as needed. It's designed to be production-ready with robust error handling, logging, and configuration options.
## Architecture
```mermaid
graph TD
A[Agent Builder] --> B[Configuration]
A --> C[Agent Creation]
A --> D[Task Execution]
B --> B1[Name]
B --> B2[Description]
B --> B3[Model Settings]
C --> C1[Agent Pool]
C --> C2[Agent Registry]
C --> C3[Agent Configuration]
D --> D1[Task Distribution]
D --> D2[Result Collection]
D --> D3[Error Handling]
C1 --> E[Specialized Agents]
C2 --> E
C3 --> E
E --> F[Task Execution]
F --> G[Results]
```
## Class Structure
### AgentsBuilder Class
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| name | str | "swarm-creator-01" | The name of the swarm |
| description | str | "This is a swarm that creates swarms" | A description of the swarm's purpose |
| verbose | bool | True | Whether to output detailed logs |
| max_loops | int | 1 | Maximum number of execution loops |
| model_name | str | "gpt-4o" | The model to use for agent creation |
| return_dictionary | bool | True | Whether to return results as a dictionary |
| system_prompt | str | BOSS_SYSTEM_PROMPT | The system prompt for the boss agent |
### Methods
| Method | Description | Parameters | Returns |
|--------|-------------|------------|---------|
| run | Run the swarm on a given task | task: str, image_url: str = None, *args, **kwargs | Tuple[List[Agent], int] |
| _create_agents | Create necessary agents for a task | task: str, *args, **kwargs | List[Agent] |
| build_agent | Build a single agent with specifications | agent_name: str, agent_description: str, agent_system_prompt: str, max_loops: int = 1, model_name: str = "gpt-4o", dynamic_temperature_enabled: bool = True, auto_generate_prompt: bool = False, role: str = "worker", max_tokens: int = 8192, temperature: float = 0.5 | Agent |
## Enterprise Use Cases
### 1. Customer Service Automation
- Create specialized agents for different aspects of customer service
- Handle ticket routing, response generation, and escalation
- Maintain consistent service quality across channels
### 2. Data Analysis Pipeline
- Build agents for data collection, cleaning, analysis, and visualization
- Automate complex data processing workflows
- Generate insights and reports automatically
### 3. Content Creation and Management
- Deploy agents for content research, writing, editing, and publishing
- Maintain brand consistency across content
- Automate content scheduling and distribution
### 4. Process Automation
- Create agents for workflow automation
- Handle document processing and routing
- Manage approval chains and notifications
### 5. Research and Development
- Build agents for literature review, experiment design, and data collection
- Automate research documentation and reporting
- Facilitate collaboration between research teams
## Example Usage
```python
from swarms import AgentsBuilder
# Initialize the agent builder
agents_builder = AgentsBuilder(
name="enterprise-automation",
description="Enterprise workflow automation swarm",
verbose=True
)
# Define a use-case for building agents
task = "Develop a swarm of agents to automate the generation of personalized marketing strategies based on customer data and market trends"
# Run the swarm
agents = agents_builder.run(task)
# Access results
print(agents)
```
## Best Practices
1. **Error Handling**
- Always implement proper error handling for agent failures
- Use retry mechanisms for transient failures
- Log all errors for debugging and monitoring
2. **Resource Management**
- Monitor agent resource usage
- Implement rate limiting for API calls
- Use connection pooling for database operations
3. **Security**
- Implement proper authentication and authorization
- Secure sensitive data and API keys
- Follow least privilege principle for agent permissions
4. **Monitoring and Logging**
- Implement comprehensive logging
- Monitor agent performance metrics
- Set up alerts for critical failures
5. **Scalability**
- Design for horizontal scaling
- Implement load balancing
- Use distributed systems when needed
## Integration Patterns
```mermaid
graph LR
A[External System] --> B[API Gateway]
B --> C[Agent Builder]
C --> D[Agent Pool]
D --> E[Specialized Agents]
E --> F[External Services]
subgraph "Monitoring"
G[Logs]
H[Metrics]
I[Alerts]
end
C --> G
C --> H
C --> I
```
## Performance Considerations
1. **Agent Pool Management**
- Implement connection pooling
- Use caching for frequently accessed data
- Optimize agent creation and destruction
2. **Task Distribution**
- Implement load balancing
- Use priority queues for task scheduling
- Handle task timeouts and retries
3. **Resource Optimization**
- Monitor memory usage
- Implement garbage collection
- Use efficient data structures
## Troubleshooting
Common issues and solutions:
1. **Agent Creation Failures**
- Check API credentials
- Verify model availability
- Review system prompts
2. **Performance Issues**
- Monitor resource usage
- Check network latency
- Review agent configurations
3. **Integration Problems**
- Verify API endpoints
- Check authentication
- Review data formats

@ -154,11 +154,6 @@ flowchart TD
- Covers detailed implementation, constructor arguments, and full examples
### Auto Agent Builder Documentation:
- [Agent Builder Documentation](https://docs.swarms.world/en/latest/swarms/structs/auto_agent_builder/)
- Includes enterprise use cases, best practices, and integration patterns
3. SwarmRouter Documentation:

@ -36,7 +36,6 @@ This page provides a comprehensive overview of all available multi-agent archite
| Architecture | Use Case | Key Functionality | Documentation |
|-------------|----------|-------------------|---------------|
| HierarchicalSwarm | Hierarchical task orchestration | Director agent coordinates specialized worker agents | [Docs](hierarchical_swarm.md) |
| Auto Agent Builder | Automated agent creation | Automatically creates and configures agents | [Docs](auto_agent_builder.md) |
| Hybrid Hierarchical-Cluster Swarm | Complex organization | Combines hierarchical and cluster-based organization | [Docs](hhcs.md) |
| Auto Swarm Builder | Automated swarm creation | Automatically creates and configures swarms | [Docs](auto_swarm_builder.md) |

@ -1,9 +0,0 @@
from swarms.structs.agent_builder import AgentsBuilder
example_task = "Write a blog post about the benefits of using Swarms for AI agents."
agents_builder = AgentsBuilder()
agents = agents_builder.run(example_task)
print(agents)

@ -0,0 +1,76 @@
BOSS_SYSTEM_PROMPT = """
# Swarm Intelligence Orchestrator
You are the Chief Orchestrator of a sophisticated agent swarm. Your primary responsibility is to analyze tasks and create the optimal team of specialized agents to accomplish complex objectives efficiently.
## Agent Creation Protocol
1. **Task Analysis**:
- Thoroughly analyze the user's task to identify all required skills, knowledge domains, and subtasks
- Break down complex problems into discrete components that can be assigned to specialized agents
- Identify potential challenges and edge cases that might require specialized handling
2. **Agent Design Principles**:
- Create highly specialized agents with clearly defined roles and responsibilities
- Design each agent with deep expertise in their specific domain
- Provide agents with comprehensive and extremely extensive system prompts that include:
* Precise definition of their role and scope of responsibility
* Detailed methodology for approaching problems in their domain
* Specific techniques, frameworks, and mental models to apply
* Guidelines for output format and quality standards
* Instructions for collaboration with other agents
* In-depth examples and scenarios to illustrate expected behavior and decision-making processes
* Extensive background information relevant to the tasks they will undertake
3. **Cognitive Enhancement**:
- Equip agents with advanced reasoning frameworks:
* First principles thinking to break down complex problems
* Systems thinking to understand interconnections
* Lateral thinking for creative solutions
* Critical thinking to evaluate information quality
- Implement specialized thought patterns:
* Step-by-step reasoning for complex problems
* Hypothesis generation and testing
* Counterfactual reasoning to explore alternatives
* Analogical reasoning to apply solutions from similar domains
4. **Swarm Architecture**:
- Design optimal agent interaction patterns based on task requirements
- Consider hierarchical, networked, or hybrid structures
- Establish clear communication protocols between agents
- Define escalation paths for handling edge cases
5. **Agent Specialization Examples**:
- Research Agents: Literature review, data gathering, information synthesis
- Analysis Agents: Data processing, pattern recognition, insight generation
- Creative Agents: Idea generation, content creation, design thinking
- Planning Agents: Strategy development, resource allocation, timeline creation
- Implementation Agents: Code writing, document drafting, execution planning
- Quality Assurance Agents: Testing, validation, error detection
- Integration Agents: Combining outputs, ensuring consistency, resolving conflicts
## Output Format
For each agent, provide:
1. **Agent Name**: Clear, descriptive title reflecting specialization
2. **Description**: Concise overview of the agent's purpose and capabilities
3. **System Prompt**: Comprehensive and extremely extensive instructions including:
- Role definition and responsibilities
- Specialized knowledge and methodologies
- Thinking frameworks and problem-solving approaches
- Output requirements and quality standards
- Collaboration guidelines with other agents
- Detailed examples and context to ensure clarity and effectiveness
## Optimization Guidelines
- Create only the agents necessary for the task - no more, no less
- Ensure each agent has a distinct, non-overlapping area of responsibility
- Design system prompts that maximize agent performance through clear guidance and specialized knowledge
- Balance specialization with the need for effective collaboration
- Prioritize agents that address the most critical aspects of the task
Remember: Your goal is to create a swarm of agents that collectively possesses the intelligence, knowledge, and capabilities to deliver exceptional results for the user's task.
"""

@ -1,5 +1,4 @@
from swarms.structs.agent import Agent
from swarms.structs.agent_builder import AgentsBuilder
from swarms.structs.agent_loader import AgentLoader
from swarms.structs.agent_rearrange import AgentRearrange, rearrange
from swarms.structs.auto_swarm_builder import AutoSwarmBuilder
@ -155,7 +154,6 @@ __all__ = [
"MultiAgentRouter",
"MemeAgentGenerator",
"ModelRouter",
"AgentsBuilder",
"MALT",
"HybridHierarchicalClusterSwarm",
"get_agents_info",

@ -1,293 +0,0 @@
import os
from typing import Any, List, Optional, Tuple
from loguru import logger
from pydantic import BaseModel, Field
from swarms.structs.agent import Agent
from swarms.utils.function_caller_model import OpenAIFunctionCaller
BOSS_SYSTEM_PROMPT = """
# Swarm Intelligence Orchestrator
You are the Chief Orchestrator of a sophisticated agent swarm. Your primary responsibility is to analyze tasks and create the optimal team of specialized agents to accomplish complex objectives efficiently.
## Agent Creation Protocol
1. **Task Analysis**:
- Thoroughly analyze the user's task to identify all required skills, knowledge domains, and subtasks
- Break down complex problems into discrete components that can be assigned to specialized agents
- Identify potential challenges and edge cases that might require specialized handling
2. **Agent Design Principles**:
- Create highly specialized agents with clearly defined roles and responsibilities
- Design each agent with deep expertise in their specific domain
- Provide agents with comprehensive and extremely extensive system prompts that include:
* Precise definition of their role and scope of responsibility
* Detailed methodology for approaching problems in their domain
* Specific techniques, frameworks, and mental models to apply
* Guidelines for output format and quality standards
* Instructions for collaboration with other agents
* In-depth examples and scenarios to illustrate expected behavior and decision-making processes
* Extensive background information relevant to the tasks they will undertake
3. **Cognitive Enhancement**:
- Equip agents with advanced reasoning frameworks:
* First principles thinking to break down complex problems
* Systems thinking to understand interconnections
* Lateral thinking for creative solutions
* Critical thinking to evaluate information quality
- Implement specialized thought patterns:
* Step-by-step reasoning for complex problems
* Hypothesis generation and testing
* Counterfactual reasoning to explore alternatives
* Analogical reasoning to apply solutions from similar domains
4. **Swarm Architecture**:
- Design optimal agent interaction patterns based on task requirements
- Consider hierarchical, networked, or hybrid structures
- Establish clear communication protocols between agents
- Define escalation paths for handling edge cases
5. **Agent Specialization Examples**:
- Research Agents: Literature review, data gathering, information synthesis
- Analysis Agents: Data processing, pattern recognition, insight generation
- Creative Agents: Idea generation, content creation, design thinking
- Planning Agents: Strategy development, resource allocation, timeline creation
- Implementation Agents: Code writing, document drafting, execution planning
- Quality Assurance Agents: Testing, validation, error detection
- Integration Agents: Combining outputs, ensuring consistency, resolving conflicts
## Output Format
For each agent, provide:
1. **Agent Name**: Clear, descriptive title reflecting specialization
2. **Description**: Concise overview of the agent's purpose and capabilities
3. **System Prompt**: Comprehensive and extremely extensive instructions including:
- Role definition and responsibilities
- Specialized knowledge and methodologies
- Thinking frameworks and problem-solving approaches
- Output requirements and quality standards
- Collaboration guidelines with other agents
- Detailed examples and context to ensure clarity and effectiveness
## Optimization Guidelines
- Create only the agents necessary for the task - no more, no less
- Ensure each agent has a distinct, non-overlapping area of responsibility
- Design system prompts that maximize agent performance through clear guidance and specialized knowledge
- Balance specialization with the need for effective collaboration
- Prioritize agents that address the most critical aspects of the task
Remember: Your goal is to create a swarm of agents that collectively possesses the intelligence, knowledge, and capabilities to deliver exceptional results for the user's task.
"""
class AgentSpec(BaseModel):
agent_name: Optional[str] = Field(
None,
description="The unique name assigned to the agent, which identifies its role and functionality within the swarm.",
)
description: Optional[str] = Field(
None,
description="A detailed explanation of the agent's purpose, capabilities, and any specific tasks it is designed to perform.",
)
system_prompt: Optional[str] = Field(
None,
description="The initial instruction or context provided to the agent, guiding its behavior and responses during execution.",
)
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"
)
auto_generate_prompt: Optional[bool] = Field(
description="A flag indicating whether the agent should automatically create prompts based on the task requirements."
)
max_tokens: Optional[int] = Field(
None,
description="The maximum number of tokens that the agent is allowed to generate in its responses, limiting output length.",
)
temperature: Optional[float] = Field(
description="A parameter that controls the randomness of the agent's output; lower values result in more deterministic responses."
)
role: Optional[str] = Field(
description="The designated role of the agent within the swarm, which influences its behavior and interaction with other agents."
)
max_loops: Optional[int] = Field(
description="The maximum number of times the agent is allowed to repeat its task, enabling iterative processing if necessary."
)
class Agents(BaseModel):
"""Configuration for a collection of agents that work together as a swarm to accomplish tasks."""
agents: List[AgentSpec] = Field(
description="A list containing the specifications of each agent that will participate in the swarm, detailing their roles and functionalities."
)
class AgentsBuilder:
"""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 = "swarm-creator-01",
description: str = "This is a swarm that creates swarms",
verbose: bool = True,
max_loops: int = 1,
model_name: str = "gpt-4o",
return_dictionary: bool = True,
system_prompt: str = BOSS_SYSTEM_PROMPT,
):
self.name = name
self.description = description
self.verbose = verbose
self.max_loops = max_loops
self.agents_pool = []
self.model_name = model_name
self.return_dictionary = return_dictionary
self.system_prompt = system_prompt
logger.info(
f"Initialized AutoSwarmBuilder: {name} {description}"
)
def run(
self, task: str, image_url: str = None, *args, **kwargs
) -> Tuple[List[Agent], int]:
"""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)
return agents
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=self.system_prompt,
api_key=os.getenv("OPENAI_API_KEY"),
temperature=0.1,
base_model=Agents,
model_name=self.model_name,
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 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(agents_dictionary)
def create_agents(self, agents_dictionary: Any):
# 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 = Agents(**agent_config)
agent = self.build_agent(
agent_name=agent_config.model_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)
return agents
def build_agent(
self,
agent_name: str,
agent_description: str,
agent_system_prompt: str,
max_loops: int = 1,
model_name: str = "gpt-4o",
dynamic_temperature_enabled: bool = True,
auto_generate_prompt: bool = False,
role: str = "worker",
max_tokens: int = 8192,
temperature: float = 0.5,
):
"""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,
model_name=model_name,
max_loops=max_loops,
dynamic_temperature_enabled=dynamic_temperature_enabled,
context_length=200000,
output_type="str", # "json", "dict", "csv" OR "string" soon "yaml" and
streaming_on=False,
auto_generate_prompt=auto_generate_prompt,
role=role,
max_tokens=max_tokens,
temperature=temperature,
)
return agent
# if __name__ == "__main__":
# builder = AgentsBuilder(model_name="gpt-4o")
# agents = builder.run("Create a swarm that can write a book about the history of the world")
# print(agents)
# print(type(agents))

@ -1,6 +1,6 @@
import json
import traceback
from typing import List, Optional
from typing import Any, List, Optional, Tuple
from dotenv import load_dotenv
from loguru import logger
@ -127,11 +127,57 @@ When creating a multi-agent system, provide:
"""
class AgentSpec(BaseModel):
"""Configuration for an individual agent specification."""
agent_name: Optional[str] = Field(
None,
description="The unique name assigned to the agent, which identifies its role and functionality within the swarm.",
)
description: Optional[str] = Field(
None,
description="A detailed explanation of the agent's purpose, capabilities, and any specific tasks it is designed to perform.",
)
system_prompt: Optional[str] = Field(
None,
description="The initial instruction or context provided to the agent, guiding its behavior and responses during execution.",
)
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"
)
auto_generate_prompt: Optional[bool] = Field(
description="A flag indicating whether the agent should automatically create prompts based on the task requirements."
)
max_tokens: Optional[int] = Field(
None,
description="The maximum number of tokens that the agent is allowed to generate in its responses, limiting output length.",
)
temperature: Optional[float] = Field(
description="A parameter that controls the randomness of the agent's output; lower values result in more deterministic responses."
)
role: Optional[str] = Field(
description="The designated role of the agent within the swarm, which influences its behavior and interaction with other agents."
)
max_loops: Optional[int] = Field(
description="The maximum number of times the agent is allowed to repeat its task, enabling iterative processing if necessary."
)
class Agents(BaseModel):
"""Configuration for a collection of agents that work together as a swarm to accomplish tasks."""
agents: List[AgentSpec] = Field(
description="A list containing the specifications of each agent that will participate in the swarm, detailing their roles and functionalities."
)
execution_types = [
"return-agents",
"execute-swarm-router",
"return-swarm-router-config",
"return-agent-configurations",
"return-agent-specs",
"return-agent-dictionary",
]
@ -225,6 +271,8 @@ class AutoSwarmBuilder:
interactive (bool): Whether to enable interactive mode. Defaults to False.
max_tokens (int): Maximum tokens for the LLM responses. Defaults to 8000.
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.
"""
def __init__(
@ -238,6 +286,8 @@ class AutoSwarmBuilder:
interactive: bool = False,
max_tokens: int = 8000,
execution_type: execution_types = "return-agents",
return_dictionary: bool = True,
system_prompt: str = BOSS_SYSTEM_PROMPT,
):
"""Initialize the AutoSwarmBuilder.
@ -251,6 +301,8 @@ class AutoSwarmBuilder:
interactive (bool): Whether to enable interactive mode
max_tokens (int): Maximum tokens for the LLM responses
execution_type (str): Type of execution to perform
return_dictionary (bool): Whether to return dictionary format for agent specs
system_prompt (str): System prompt for the boss agent
"""
self.name = name
self.description = description
@ -261,7 +313,10 @@ class AutoSwarmBuilder:
self.interactive = interactive
self.max_tokens = max_tokens
self.execution_type = execution_type
self.return_dictionary = return_dictionary
self.system_prompt = system_prompt
self.conversation = Conversation()
self.agents_pool = []
self.reliability_check()
@ -307,6 +362,10 @@ class AutoSwarmBuilder:
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)
@ -358,14 +417,6 @@ class AutoSwarmBuilder:
raise e
def build_llm_agent(self, config: BaseModel):
# return OpenAIFunctionCaller(
# system_prompt=BOSS_SYSTEM_PROMPT,
# api_key=os.getenv("OPENAI_API_KEY"),
# temperature=0.5,
# base_model=config,
# model_name=self.model_name,
# max_tokens=self.max_tokens,
# )
return LiteLLM(
model_name=self.model_name,
system_prompt=BOSS_SYSTEM_PROMPT,
@ -504,5 +555,146 @@ class AutoSwarmBuilder:
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(self, agents_dictionary: Any) -> List[Agent]:
"""Create agents from agent specifications.
Args:
agents_dictionary: Dictionary containing agent specifications
Returns:
List[Agent]: List of created agents
"""
# Create agents from config
agents = []
for agent_config in agents_dictionary.agents:
# Convert dict to AgentSpec if needed
if isinstance(agent_config, dict):
agent_config = AgentSpec(**agent_config)
agent = self.build_agent_from_spec(
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)
return agents
def build_agent_from_spec(
self,
agent_name: str,
agent_description: str,
agent_system_prompt: str,
max_loops: int = 1,
model_name: str = "gpt-4o",
dynamic_temperature_enabled: bool = True,
auto_generate_prompt: bool = False,
role: str = "worker",
max_tokens: int = 8192,
temperature: float = 0.5,
) -> Agent:
"""Build a single agent from agent specification.
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
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:
Agent: The constructed agent instance
"""
logger.info(f"Building agent from spec: {agent_name}")
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):
return execution_types

@ -496,6 +496,7 @@ class LiteLLM:
if isinstance(out, BaseModel):
out = out.model_dump()
return out
def output_for_reasoning(self, response: any):

Loading…
Cancel
Save