fix system prompt, fix agent prompting swarms, new auto swarm builder, auto swarm builder docs, and more
Kye Gomez
1 week ago
parent
0948fdff43
commit
cc387abaf6
@ -0,0 +1,17 @@
|
||||
from swarms.structs.auto_swarm_builder import AutoSwarmBuilder
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
swarm = AutoSwarmBuilder(
|
||||
name="My Swarm",
|
||||
description="My Swarm Description",
|
||||
verbose=True,
|
||||
max_loops=1,
|
||||
)
|
||||
|
||||
result = swarm.run(
|
||||
task="Build a swarm to write a research paper on the topic of AI"
|
||||
)
|
||||
|
||||
print(result)
|
||||
@ -0,0 +1,14 @@
|
||||
from swarms.structs.auto_swarm_builder import AutoSwarmBuilder
|
||||
|
||||
|
||||
example = AutoSwarmBuilder(
|
||||
name="ChipDesign-Swarm",
|
||||
description="A swarm of specialized AI agents collaborating on chip architecture, logic design, verification, and optimization to create novel semiconductor designs",
|
||||
max_loops=1,
|
||||
)
|
||||
|
||||
print(
|
||||
example.run(
|
||||
"Design a new AI accelerator chip optimized for transformer model inference. Consider the following aspects: 1) Overall chip architecture and block diagram 2) Memory hierarchy and interconnects 3) Processing elements and data flow 4) Power and thermal considerations 5) Physical layout recommendations -> "
|
||||
)
|
||||
)
|
||||
@ -0,0 +1,643 @@
|
||||
"""
|
||||
CEO -> Finds department leader
|
||||
Department leader -> Finds employees
|
||||
Employees -> Do the work
|
||||
|
||||
Todo
|
||||
- Create schemas that enable the ceo to find the department leader or leaders
|
||||
- CEO then distributes orders to department leaders or just one leader
|
||||
- Department leader then distributes orders to employees
|
||||
- Employees can choose to do the work or delegate to another employee or work together
|
||||
- When the employees are done, they report back to the department leader
|
||||
- Department leader then reports back to the ceo
|
||||
- CEO then reports back to the user
|
||||
|
||||
|
||||
|
||||
Logic
|
||||
- dynamically setup conversations for each department -- Feed context to each agent in the department
|
||||
- Feed context to each agent in the department
|
||||
"""
|
||||
|
||||
from typing import Callable, List, Union
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from swarms.structs.agent import Agent
|
||||
from swarms.structs.conversation import Conversation
|
||||
from swarms.structs.ma_utils import list_all_agents
|
||||
from swarms.utils.str_to_dict import str_to_dict
|
||||
from swarms.utils.any_to_str import any_to_str
|
||||
|
||||
|
||||
class Department(BaseModel):
|
||||
name: str = Field(description="The name of the department")
|
||||
description: str = Field(
|
||||
description="A description of the department"
|
||||
)
|
||||
employees: List[Union[Agent, Callable]] = Field(
|
||||
description="A list of employees in the department"
|
||||
)
|
||||
leader_name: str = Field(
|
||||
description="The name of the leader of the department"
|
||||
)
|
||||
|
||||
class Config:
|
||||
arbitrary_types_allowed = True
|
||||
|
||||
|
||||
CEO_SCHEMA = {
|
||||
"name": "delegate_task_to_department",
|
||||
"description": "CEO function to analyze and delegate tasks to appropriate department leaders",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"thought": {
|
||||
"type": "string",
|
||||
"description": "Reasoning about the task, its requirements, and potential approaches",
|
||||
},
|
||||
"plan": {
|
||||
"type": "string",
|
||||
"description": "Structured plan for how to accomplish the task across departments",
|
||||
},
|
||||
"tasks": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"task_description": {"type": "string"},
|
||||
"selected_departments": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": "List of department names that should handle this task",
|
||||
},
|
||||
"selected_leaders": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": "List of department leaders to assign the task to",
|
||||
},
|
||||
"success_criteria": {"type": "string"},
|
||||
},
|
||||
"required": [
|
||||
"task_description",
|
||||
"selected_departments",
|
||||
"selected_leaders",
|
||||
],
|
||||
},
|
||||
},
|
||||
"required": ["thought", "plan", "tasks"],
|
||||
},
|
||||
}
|
||||
|
||||
DEPARTMENT_LEADER_SCHEMA = {
|
||||
"name": "manage_department_task",
|
||||
"description": "Department leader function to break down and assign tasks to employees",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"task_management": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"original_task": {"type": "string"},
|
||||
"subtasks": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"subtask_id": {"type": "string"},
|
||||
"description": {"type": "string"},
|
||||
"assigned_employees": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
"estimated_duration": {
|
||||
"type": "string"
|
||||
},
|
||||
"dependencies": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
"progress_tracking": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"status": {
|
||||
"type": "string",
|
||||
"enum": [
|
||||
"not_started",
|
||||
"in_progress",
|
||||
"completed",
|
||||
],
|
||||
},
|
||||
"completion_percentage": {
|
||||
"type": "number"
|
||||
},
|
||||
"blockers": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
"required": ["original_task", "subtasks"],
|
||||
}
|
||||
},
|
||||
"required": ["task_management"],
|
||||
},
|
||||
}
|
||||
|
||||
EMPLOYEE_SCHEMA = {
|
||||
"name": "handle_assigned_task",
|
||||
"description": "Employee function to process and execute assigned tasks",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"thought": {
|
||||
"type": "string",
|
||||
"description": "Reasoning about the task, its requirements, and potential approaches",
|
||||
},
|
||||
"plan": {
|
||||
"type": "string",
|
||||
"description": "Structured plan for how to accomplish the task across departments",
|
||||
},
|
||||
"task_execution": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"subtask_id": {"type": "string"},
|
||||
"action_taken": {
|
||||
"type": "string",
|
||||
"enum": [
|
||||
"execute",
|
||||
"delegate",
|
||||
"collaborate",
|
||||
],
|
||||
},
|
||||
"execution_details": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"status": {
|
||||
"type": "string",
|
||||
"enum": [
|
||||
"in_progress",
|
||||
"completed",
|
||||
"blocked",
|
||||
],
|
||||
},
|
||||
"work_log": {"type": "string"},
|
||||
"collaboration_partners": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
"delegate_to": {"type": "string"},
|
||||
"results": {"type": "string"},
|
||||
"issues_encountered": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
"required": [
|
||||
"thought",
|
||||
"plan",
|
||||
"subtask_id",
|
||||
"action_taken",
|
||||
"execution_details",
|
||||
],
|
||||
},
|
||||
},
|
||||
"required": ["task_execution"],
|
||||
},
|
||||
}
|
||||
|
||||
# Status report schemas for the feedback loop
|
||||
EMPLOYEE_REPORT_SCHEMA = {
|
||||
"name": "submit_task_report",
|
||||
"description": "Employee function to report task completion status to department leader",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"task_report": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"subtask_id": {"type": "string"},
|
||||
"completion_status": {
|
||||
"type": "string",
|
||||
"enum": ["completed", "partial", "blocked"],
|
||||
},
|
||||
"work_summary": {"type": "string"},
|
||||
"time_spent": {"type": "string"},
|
||||
"challenges": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
"next_steps": {"type": "string"},
|
||||
},
|
||||
"required": [
|
||||
"subtask_id",
|
||||
"completion_status",
|
||||
"work_summary",
|
||||
],
|
||||
}
|
||||
},
|
||||
"required": ["task_report"],
|
||||
},
|
||||
}
|
||||
|
||||
DEPARTMENT_REPORT_SCHEMA = {
|
||||
"name": "submit_department_report",
|
||||
"description": "Department leader function to report department progress to CEO",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"department_report": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"department_name": {"type": "string"},
|
||||
"task_summary": {"type": "string"},
|
||||
"overall_status": {
|
||||
"type": "string",
|
||||
"enum": ["on_track", "at_risk", "completed"],
|
||||
},
|
||||
"completion_percentage": {"type": "number"},
|
||||
"key_achievements": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
"blockers": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
"resource_needs": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
"next_milestones": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
},
|
||||
"required": [
|
||||
"department_name",
|
||||
"task_summary",
|
||||
"overall_status",
|
||||
],
|
||||
}
|
||||
},
|
||||
"required": ["department_report"],
|
||||
},
|
||||
}
|
||||
|
||||
CEO_FINAL_REPORT_SCHEMA = {
|
||||
"name": "generate_final_report",
|
||||
"description": "CEO function to compile final report for the user",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"final_report": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"task_overview": {"type": "string"},
|
||||
"overall_status": {
|
||||
"type": "string",
|
||||
"enum": ["successful", "partial", "failed"],
|
||||
},
|
||||
"department_summaries": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"department": {"type": "string"},
|
||||
"contribution": {"type": "string"},
|
||||
"performance": {"type": "string"},
|
||||
},
|
||||
},
|
||||
},
|
||||
"final_results": {"type": "string"},
|
||||
"recommendations": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
"next_steps": {"type": "string"},
|
||||
},
|
||||
"required": [
|
||||
"task_overview",
|
||||
"overall_status",
|
||||
"final_results",
|
||||
],
|
||||
}
|
||||
},
|
||||
"required": ["final_report"],
|
||||
},
|
||||
}
|
||||
|
||||
# # Example output schemas
|
||||
# CEO_EXAMPLE_OUTPUT = {
|
||||
# "thought": "This task requires coordination between the engineering and design departments to create a new feature. The engineering team will handle the backend implementation while design focuses on the user interface.",
|
||||
# "plan": "1. Assign backend development to engineering department\n2. Assign UI/UX design to design department\n3. Set up regular sync meetings between departments\n4. Establish clear success criteria",
|
||||
# "tasks": {
|
||||
# "task_description": "Develop a new user authentication system with social login integration",
|
||||
# "selected_departments": ["engineering", "design"],
|
||||
# "selected_leaders": ["engineering_lead", "design_lead"],
|
||||
# "success_criteria": "1. Social login working with 3 major providers\n2. UI/UX approved by design team\n3. Security audit passed\n4. Performance metrics met"
|
||||
# }
|
||||
# }
|
||||
|
||||
# DEPARTMENT_LEADER_EXAMPLE_OUTPUT = {
|
||||
# "task_management": {
|
||||
# "original_task": "Develop a new user authentication system with social login integration",
|
||||
# "subtasks": [
|
||||
# {
|
||||
# "subtask_id": "ENG-001",
|
||||
# "description": "Implement OAuth2 integration for Google",
|
||||
# "assigned_employees": ["dev1", "dev2"],
|
||||
# "estimated_duration": "3 days",
|
||||
# "dependencies": ["DES-001"]
|
||||
# },
|
||||
# {
|
||||
# "subtask_id": "ENG-002",
|
||||
# "description": "Implement OAuth2 integration for Facebook",
|
||||
# "assigned_employees": ["dev3"],
|
||||
# "estimated_duration": "2 days",
|
||||
# "dependencies": ["DES-001"]
|
||||
# }
|
||||
# ],
|
||||
# "progress_tracking": {
|
||||
# "status": "in_progress",
|
||||
# "completion_percentage": 0.3,
|
||||
# "blockers": ["Waiting for design team to provide UI mockups"]
|
||||
# }
|
||||
# }
|
||||
# }
|
||||
|
||||
# EMPLOYEE_EXAMPLE_OUTPUT = {
|
||||
# "thought": "The Google OAuth2 integration requires careful handling of token management and user data synchronization",
|
||||
# "plan": "1. Set up Google OAuth2 credentials\n2. Implement token refresh mechanism\n3. Create user data sync pipeline\n4. Add error handling and logging",
|
||||
# "task_execution": {
|
||||
# "subtask_id": "ENG-001",
|
||||
# "action_taken": "execute",
|
||||
# "execution_details": {
|
||||
# "status": "in_progress",
|
||||
# "work_log": "Completed OAuth2 credential setup and initial token handling implementation",
|
||||
# "collaboration_partners": ["dev2"],
|
||||
# "delegate_to": None,
|
||||
# "results": "Successfully implemented basic OAuth2 flow",
|
||||
# "issues_encountered": ["Need to handle token refresh edge cases"]
|
||||
# }
|
||||
# }
|
||||
# }
|
||||
|
||||
# EMPLOYEE_REPORT_EXAMPLE = {
|
||||
# "task_report": {
|
||||
# "subtask_id": "ENG-001",
|
||||
# "completion_status": "partial",
|
||||
# "work_summary": "Completed initial OAuth2 implementation, working on token refresh mechanism",
|
||||
# "time_spent": "2 days",
|
||||
# "challenges": ["Token refresh edge cases", "Rate limiting considerations"],
|
||||
# "next_steps": "Implement token refresh mechanism and add rate limiting protection"
|
||||
# }
|
||||
# }
|
||||
|
||||
# DEPARTMENT_REPORT_EXAMPLE = {
|
||||
# "department_report": {
|
||||
# "department_name": "Engineering",
|
||||
# "task_summary": "Making good progress on OAuth2 implementation, but waiting on design team for UI components",
|
||||
# "overall_status": "on_track",
|
||||
# "completion_percentage": 0.4,
|
||||
# "key_achievements": [
|
||||
# "Completed Google OAuth2 basic flow",
|
||||
# "Set up secure token storage"
|
||||
# ],
|
||||
# "blockers": ["Waiting for UI mockups from design team"],
|
||||
# "resource_needs": ["Additional QA resources for testing"],
|
||||
# "next_milestones": [
|
||||
# "Complete Facebook OAuth2 integration",
|
||||
# "Implement token refresh mechanism"
|
||||
# ]
|
||||
# }
|
||||
# }
|
||||
|
||||
# CEO_FINAL_REPORT_EXAMPLE = {
|
||||
# "final_report": {
|
||||
# "task_overview": "Successfully implemented new authentication system with social login capabilities",
|
||||
# "overall_status": "successful",
|
||||
# "department_summaries": [
|
||||
# {
|
||||
# "department": "Engineering",
|
||||
# "contribution": "Implemented secure OAuth2 integrations and token management",
|
||||
# "performance": "Excellent - completed all technical requirements"
|
||||
# },
|
||||
# {
|
||||
# "department": "Design",
|
||||
# "contribution": "Created intuitive UI/UX for authentication flows",
|
||||
# "performance": "Good - delivered all required designs on time"
|
||||
# }
|
||||
# ],
|
||||
# "final_results": "New authentication system is live and processing 1000+ logins per day",
|
||||
# "recommendations": [
|
||||
# "Add more social login providers",
|
||||
# "Implement biometric authentication",
|
||||
# "Add two-factor authentication"
|
||||
# ],
|
||||
# "next_steps": "Monitor system performance and gather user feedback for improvements"
|
||||
# }
|
||||
# }
|
||||
|
||||
|
||||
class AutoCorp:
|
||||
def __init__(
|
||||
self,
|
||||
name: str = "AutoCorp",
|
||||
description: str = "A company that uses agents to automate tasks",
|
||||
departments: List[Department] = [],
|
||||
ceo: Agent = None,
|
||||
):
|
||||
self.name = name
|
||||
self.description = description
|
||||
self.departments = departments
|
||||
self.ceo = ceo
|
||||
self.conversation = Conversation()
|
||||
|
||||
# Check if the CEO and departments are set
|
||||
self.reliability_check()
|
||||
|
||||
# Add departments to conversation
|
||||
self.add_departments_to_conversation()
|
||||
|
||||
# Initialize the CEO agent
|
||||
self.initialize_ceo_agent()
|
||||
|
||||
# Initialize the department leaders
|
||||
self.setup_department_leaders()
|
||||
|
||||
# Initialize the department employees
|
||||
self.department_employees_initialize()
|
||||
|
||||
def initialize_ceo_agent(self):
|
||||
self.ceo.tools_list_dictionary = [
|
||||
CEO_SCHEMA,
|
||||
CEO_FINAL_REPORT_SCHEMA,
|
||||
]
|
||||
|
||||
def setup_department_leaders(self):
|
||||
self.department_leader_initialize()
|
||||
self.initialize_department_leaders()
|
||||
|
||||
def department_leader_initialize(self):
|
||||
"""Initialize each department leader with their department's context."""
|
||||
|
||||
for department in self.departments:
|
||||
# Create a context dictionary for the department
|
||||
department_context = {
|
||||
"name": department.name,
|
||||
"description": department.description,
|
||||
"employees": list_all_agents(
|
||||
department.employees,
|
||||
self.conversation,
|
||||
department.name,
|
||||
False,
|
||||
),
|
||||
}
|
||||
|
||||
# Convert the context to a string
|
||||
context_str = any_to_str(department_context)
|
||||
|
||||
# TODO: Add the department leader's tools and context
|
||||
department.leader.system_prompt += f"""
|
||||
You are the leader of the {department.name} department.
|
||||
|
||||
Department Context:
|
||||
{context_str}
|
||||
|
||||
Your role is to:
|
||||
1. Break down tasks into subtasks
|
||||
2. Assign subtasks to appropriate employees
|
||||
3. Track progress and manage blockers
|
||||
4. Report back to the CEO
|
||||
|
||||
Use the provided tools to manage your department effectively.
|
||||
"""
|
||||
|
||||
def department_employees_initialize(self):
|
||||
"""Initialize each department leader with their department's context."""
|
||||
|
||||
for department in self.departments:
|
||||
# Create a context dictionary for the department
|
||||
department_context = {
|
||||
"name": department.name,
|
||||
"description": department.description,
|
||||
"employees": list_all_agents(
|
||||
department.employees,
|
||||
self.conversation,
|
||||
department.name,
|
||||
False,
|
||||
),
|
||||
"leader": department.leader_name,
|
||||
}
|
||||
|
||||
print(department_context)
|
||||
|
||||
# Convert the context to a string
|
||||
context_str = any_to_str(department_context)
|
||||
|
||||
# Set the department leader's tools and context
|
||||
department.employees.system_prompt += f"""
|
||||
You are an employee of the {department.name} department.
|
||||
|
||||
Department Context:
|
||||
{context_str}
|
||||
|
||||
Your role is to:
|
||||
1. Break down tasks into subtasks
|
||||
2. Assign subtasks to appropriate employees
|
||||
3. Track progress and manage blockers
|
||||
4. Report back to the CEO
|
||||
|
||||
Use the provided tools to manage your department effectively.
|
||||
"""
|
||||
|
||||
def initialize_department_leaders(self):
|
||||
# Use list comprehension for faster initialization
|
||||
[
|
||||
setattr(
|
||||
dept.leader,
|
||||
"tools_list_dictionary",
|
||||
[DEPARTMENT_LEADER_SCHEMA],
|
||||
)
|
||||
for dept in self.departments
|
||||
]
|
||||
|
||||
def reliability_check(self):
|
||||
if self.ceo is None:
|
||||
raise ValueError("CEO is not set")
|
||||
|
||||
if self.departments is None:
|
||||
raise ValueError("No departments are set")
|
||||
|
||||
if len(self.departments) == 0:
|
||||
raise ValueError("No departments are set")
|
||||
|
||||
def add_departments_to_conversation(self):
|
||||
# Batch process departments using list comprehension
|
||||
messages = [
|
||||
{
|
||||
"role": "System",
|
||||
"content": f"Team: {dept.name}\nDescription: {dept.description}\nLeader: {dept.leader_name}\nAgents: {list_all_agents(dept.employees, self.conversation, dept.name, False)}",
|
||||
}
|
||||
for dept in self.departments
|
||||
]
|
||||
self.conversation.batch_add(messages)
|
||||
|
||||
# def add_department(self, department: Department):
|
||||
# self.departments.append(department)
|
||||
|
||||
# def add_employee(self, employee: Union[Agent, Callable]):
|
||||
# self.departments[-1].employees.append(employee)
|
||||
|
||||
# def add_ceo(self, ceo: Agent):
|
||||
# self.ceo = ceo
|
||||
|
||||
# def add_employee_to_department(
|
||||
# self, employee: Union[Agent, Callable], department: Department
|
||||
# ):
|
||||
# department.employees.append(employee)
|
||||
|
||||
# def add_leader_to_department(
|
||||
# self, leader: Agent, department: Department
|
||||
# ):
|
||||
# department.leader = leader
|
||||
|
||||
# def add_department_to_auto_corp(self, department: Department):
|
||||
# self.departments.append(department)
|
||||
|
||||
# def add_ceo_to_auto_corp(self, ceo: Agent):
|
||||
# self.ceo = ceo
|
||||
|
||||
# def add_employee_to_ceo(self, employee: Union[Agent, Callable]):
|
||||
# self.ceo.employees.append(employee)
|
||||
|
||||
def run(self, task: str):
|
||||
self.ceo_to_department_leaders(task)
|
||||
|
||||
# Then the department leaders to employees
|
||||
|
||||
def ceo_to_department_leaders(self, task: str):
|
||||
orders = self.ceo.run(
|
||||
f"History: {self.conversation.get_str()}\n Your Current Task: {task}"
|
||||
)
|
||||
|
||||
orders = str_to_dict(orders)
|
||||
|
||||
for department in orders["tasks"]["selected_departments"]:
|
||||
department_leader = self.departments[department].leader
|
||||
|
||||
# Get the department leader to break down the task
|
||||
outputs = department_leader.run(
|
||||
orders["tasks"]["selected_leaders"]
|
||||
)
|
||||
|
||||
# Add the department leader's response to the conversation
|
||||
self.conversation.add(
|
||||
role=f"{department_leader.name} from {department}",
|
||||
content=outputs,
|
||||
)
|
||||
@ -0,0 +1,179 @@
|
||||
# AutoSwarmBuilder Documentation
|
||||
|
||||
The `AutoSwarmBuilder` is a powerful class that automatically builds and manages swarms of AI agents to accomplish complex tasks. It uses a boss agent to delegate work and create specialized agents as needed.
|
||||
|
||||
## Overview
|
||||
|
||||
The AutoSwarmBuilder is designed to:
|
||||
|
||||
- Automatically create and coordinate multiple AI agents
|
||||
|
||||
- Delegate tasks to specialized agents
|
||||
|
||||
- Manage communication between agents
|
||||
|
||||
- Handle complex workflows through a swarm router
|
||||
|
||||
|
||||
## Parameters
|
||||
|
||||
| Parameter | Type | Default | Description |
|
||||
|-----------|------|---------|-------------|
|
||||
| name | str | None | The name of the swarm |
|
||||
| description | str | None | A description of the swarm's purpose |
|
||||
| verbose | bool | True | Whether to output detailed logs |
|
||||
| max_loops | int | 1 | Maximum number of execution loops |
|
||||
| random_models | bool | True | Whether to use random models for agents |
|
||||
|
||||
## Core Methods
|
||||
|
||||
### run(task: str, *args, **kwargs)
|
||||
|
||||
Executes the swarm on a given task.
|
||||
|
||||
**Parameters:**
|
||||
|
||||
- `task` (str): The task to execute
|
||||
|
||||
- `*args`: Additional positional arguments
|
||||
|
||||
- `**kwargs`: Additional keyword arguments
|
||||
|
||||
**Returns:**
|
||||
|
||||
- The result of the swarm execution
|
||||
|
||||
### create_agents(task: str)
|
||||
|
||||
Creates specialized agents for a given task.
|
||||
|
||||
**Parameters:**
|
||||
|
||||
- `task` (str): The task to create agents for
|
||||
|
||||
**Returns:**
|
||||
|
||||
- List[Agent]: List of created agents
|
||||
|
||||
### build_agent(agent_name: str, agent_description: str, agent_system_prompt: str)
|
||||
Builds a single agent with specified parameters.
|
||||
|
||||
**Parameters:**
|
||||
- `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
|
||||
|
||||
### batch_run(tasks: List[str])
|
||||
|
||||
Executes the swarm on multiple tasks.
|
||||
|
||||
**Parameters:**
|
||||
|
||||
- `tasks` (List[str]): List of tasks to execute
|
||||
|
||||
**Returns:**
|
||||
|
||||
- List[Any]: Results from each task execution
|
||||
|
||||
## Examples
|
||||
|
||||
### Example 1: Content Creation Swarm
|
||||
|
||||
```python
|
||||
from swarms.structs.auto_swarm_builder import AutoSwarmBuilder
|
||||
|
||||
# Initialize the swarm builder
|
||||
swarm = AutoSwarmBuilder(
|
||||
name="Content Creation Swarm",
|
||||
description="A swarm specialized in creating high-quality content"
|
||||
)
|
||||
|
||||
# Run the swarm on a content creation task
|
||||
result = swarm.run(
|
||||
"Create a comprehensive blog post about artificial intelligence in healthcare, "
|
||||
"including current applications, future trends, and ethical considerations."
|
||||
)
|
||||
```
|
||||
|
||||
### Example 2: Data Analysis Swarm
|
||||
|
||||
```python
|
||||
from swarms.structs.auto_swarm_builder import AutoSwarmBuilder
|
||||
|
||||
# Initialize the swarm builder
|
||||
swarm = AutoSwarmBuilder(
|
||||
name="Data Analysis Swarm",
|
||||
description="A swarm specialized in data analysis and visualization"
|
||||
)
|
||||
|
||||
# Run the swarm on a data analysis task
|
||||
result = swarm.run(
|
||||
"Analyze the provided sales data and create a detailed report with visualizations "
|
||||
"showing trends, patterns, and recommendations for improvement."
|
||||
)
|
||||
```
|
||||
|
||||
### Example 3: Batch Processing Multiple Tasks
|
||||
|
||||
```python
|
||||
from swarms.structs.auto_swarm_builder import AutoSwarmBuilder
|
||||
|
||||
# Initialize the swarm builder
|
||||
swarm = AutoSwarmBuilder(
|
||||
name="Multi-Task Swarm",
|
||||
description="A swarm capable of handling multiple diverse tasks"
|
||||
)
|
||||
|
||||
# Define multiple tasks
|
||||
tasks = [
|
||||
"Create a marketing strategy for a new product launch",
|
||||
"Analyze customer feedback and generate improvement suggestions",
|
||||
"Develop a project timeline for the next quarter"
|
||||
]
|
||||
|
||||
# Run the swarm on all tasks
|
||||
results = swarm.batch_run(tasks)
|
||||
```
|
||||
|
||||
## Best Practices
|
||||
|
||||
!!! tip "Task Definition"
|
||||
- Provide clear, specific task descriptions
|
||||
|
||||
- Include any relevant context or constraints
|
||||
|
||||
- Specify expected output format if needed
|
||||
|
||||
!!! note "Configuration"
|
||||
|
||||
- Set appropriate `max_loops` based on task complexity
|
||||
|
||||
- Use `verbose=True` during development for debugging
|
||||
|
||||
- Consider using `random_models=True` for diverse agent capabilities
|
||||
|
||||
!!! warning "Error Handling"
|
||||
- The class includes comprehensive error handling
|
||||
|
||||
- All methods include try-catch blocks with detailed logging
|
||||
|
||||
- Errors are propagated with full stack traces for debugging
|
||||
|
||||
## Notes
|
||||
|
||||
!!! info "Architecture"
|
||||
|
||||
- The AutoSwarmBuilder uses a sophisticated boss agent system to coordinate tasks
|
||||
|
||||
- Agents are created dynamically based on task requirements
|
||||
|
||||
- The system includes built-in logging and error handling
|
||||
|
||||
- Results are returned in a structured format for easy processing
|
||||
@ -1,196 +0,0 @@
|
||||
# Swarms Telemetry API Documentation
|
||||
|
||||
This documentation covers the API for handling telemetry data. The API is implemented using Next.js, Supabase for data storage, and Zod for request validation. The handler processes incoming telemetry data, validates it, and stores it in a Supabase database. The handler also includes robust error handling and retries for database insertions to ensure data reliability.
|
||||
|
||||
## Endpoint
|
||||
|
||||
- **URL:** `/api/telemetry`
|
||||
- **Method:** `POST`
|
||||
- **Content-Type:** `application/json`
|
||||
- **Description:** Receives telemetry data and stores it in the Supabase database.
|
||||
|
||||
## Request Schema
|
||||
|
||||
The API expects a JSON object in the request body that matches the following schema, validated using Zod:
|
||||
|
||||
| Field Name | Type | Required | Description |
|
||||
|---------------------|----------|----------|-----------------------------------------------------------|
|
||||
| `data` | `any` | No | Telemetry data payload. |
|
||||
| `swarms_api_key` | `string` | No | API key associated with the swarms framework. |
|
||||
| `status` | `string` | No | Status of the telemetry data. Default is `'received'`. |
|
||||
| `processing_time` | `string` | No | Time taken to process the telemetry data. |
|
||||
|
||||
## Response
|
||||
|
||||
### Success Response
|
||||
|
||||
- **Status Code:** `200 OK`
|
||||
- **Content-Type:** `application/json`
|
||||
- **Body:**
|
||||
|
||||
```json
|
||||
{
|
||||
"message": "Telemetry data received and stored successfully"
|
||||
}
|
||||
```
|
||||
|
||||
### Error Responses
|
||||
|
||||
- **Status Code:** `400 Bad Request`
|
||||
- **Content-Type:** `application/json`
|
||||
- **Body:**
|
||||
|
||||
```json
|
||||
{
|
||||
"error": "Invalid data format",
|
||||
"details": [
|
||||
// Zod validation error details
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
- **Status Code:** `405 Method Not Allowed`
|
||||
- **Content-Type:** `application/json`
|
||||
- **Body:**
|
||||
|
||||
```json
|
||||
{
|
||||
"error": "Method Not Allowed"
|
||||
}
|
||||
```
|
||||
|
||||
- **Status Code:** `500 Internal Server Error`
|
||||
- **Content-Type:** `application/json`
|
||||
- **Body:**
|
||||
|
||||
```json
|
||||
{
|
||||
"error": "Internal Server Error",
|
||||
"details": "Error message"
|
||||
}
|
||||
```
|
||||
|
||||
## Example Usage
|
||||
|
||||
### Python (Using `requests` Library)
|
||||
|
||||
```python
|
||||
import requests
|
||||
|
||||
url = "https://swarms.world/api/telemetry"
|
||||
headers = {
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
data = {
|
||||
"data": {"example_key": "example_value"},
|
||||
"swarms_api_key": "your_swarms_api_key",
|
||||
"status": "received",
|
||||
"processing_time": "123ms"
|
||||
}
|
||||
|
||||
response = requests.post(url, json=data, headers=headers)
|
||||
|
||||
print(response.status_code)
|
||||
print(response.json())
|
||||
```
|
||||
|
||||
### Node.js (Using `axios` Library)
|
||||
|
||||
```javascript
|
||||
const axios = require('axios');
|
||||
|
||||
const url = 'https://swarms.world/api/telemetry';
|
||||
const data = {
|
||||
data: { example_key: 'example_value' },
|
||||
swarms_api_key: 'your_swarms_api_key',
|
||||
status: 'received',
|
||||
processing_time: '123ms'
|
||||
};
|
||||
|
||||
axios.post(url, data)
|
||||
.then(response => {
|
||||
console.log(response.status);
|
||||
console.log(response.data);
|
||||
})
|
||||
.catch(error => {
|
||||
console.error(error.response.status);
|
||||
console.error(error.response.data);
|
||||
});
|
||||
```
|
||||
|
||||
### Go (Using `net/http` and `encoding/json`)
|
||||
|
||||
```go
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
)
|
||||
|
||||
func main() {
|
||||
url := "https://swarms.world/api/telemetry"
|
||||
data := map[string]interface{}{
|
||||
"data": map[string]interface{}{"example_key": "example_value"},
|
||||
"swarms_api_key": "your_swarms_api_key",
|
||||
"status": "received",
|
||||
"processing_time": "123ms",
|
||||
}
|
||||
jsonData, err := json.Marshal(data)
|
||||
if err != nil {
|
||||
fmt.Println("Error marshaling JSON:", err)
|
||||
return
|
||||
}
|
||||
|
||||
req, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonData))
|
||||
if err != nil {
|
||||
fmt.Println("Error creating request:", err)
|
||||
return
|
||||
}
|
||||
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
client := &http.Client{}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
fmt.Println("Error making request:", err)
|
||||
return
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
fmt.Println("Response status:", resp.Status)
|
||||
}
|
||||
```
|
||||
|
||||
### cURL Command
|
||||
|
||||
```bash
|
||||
curl -X POST https://swarms.world/api/telemetry \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{
|
||||
"data": {"example_key": "example_value"},
|
||||
"swarms_api_key": "your_swarms_api_key",
|
||||
"status": "received",
|
||||
"processing_time": "123ms"
|
||||
}'
|
||||
```
|
||||
|
||||
### Supabase Table Structure
|
||||
|
||||
The Supabase table (presumably `swarms_framework_schema`) should have the following columns:
|
||||
|
||||
- **`data`**: JSONB or TEXT - Stores the telemetry data payload.
|
||||
- **`swarms_api_key`**: TEXT - Stores the API key associated with the data.
|
||||
- **`source_ip`**: TEXT - Stores the IP address of the request source.
|
||||
- **`status`**: TEXT - Stores the status of the data processing.
|
||||
- **`processing_time`**: TEXT - Stores the time taken to process the telemetry data.
|
||||
|
||||
## References and Further Reading
|
||||
|
||||
- [Next.js API Routes Documentation](https://nextjs.org/docs/api-routes/introduction)
|
||||
- [Supabase JavaScript Client](https://supabase.com/docs/reference/javascript/supabase-client)
|
||||
- [Zod Schema Validation](https://zod.dev/)
|
||||
- [OpenAPI Specification](https://swagger.io/specification/)
|
||||
|
||||
This documentation is designed to be thorough and provide all the necessary details for developers to effectively use and integrate with the telemetry API.
|
||||
@ -1,68 +0,0 @@
|
||||
import os
|
||||
|
||||
from swarms import Agent
|
||||
|
||||
from swarm_models import OpenAIChat
|
||||
from swarms.structs.agents_available import showcase_available_agents
|
||||
|
||||
# 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(
|
||||
api_key=api_key, model_name="gpt-4o-mini", temperature=0.1
|
||||
)
|
||||
|
||||
# Initialize the Claims Director agent
|
||||
director_agent = Agent(
|
||||
agent_name="ClaimsDirector",
|
||||
agent_description="Oversees and coordinates the medical insurance claims processing workflow",
|
||||
system_prompt="""You are the Claims Director responsible for managing the medical insurance claims process.
|
||||
Assign and prioritize tasks between claims processors and auditors. Ensure claims are handled efficiently
|
||||
and accurately while maintaining compliance with insurance policies and regulations.""",
|
||||
llm=model,
|
||||
max_loops=1,
|
||||
dashboard=False,
|
||||
streaming_on=True,
|
||||
verbose=True,
|
||||
stopping_token="<DONE>",
|
||||
state_save_file_type="json",
|
||||
saved_state_path="director_agent.json",
|
||||
)
|
||||
|
||||
# Initialize Claims Processor agent
|
||||
processor_agent = Agent(
|
||||
agent_name="ClaimsProcessor",
|
||||
agent_description="Reviews and processes medical insurance claims, verifying coverage and eligibility",
|
||||
system_prompt="""Review medical insurance claims for completeness and accuracy. Verify patient eligibility,
|
||||
coverage details, and process claims according to policy guidelines. Flag any claims requiring special review.""",
|
||||
llm=model,
|
||||
max_loops=1,
|
||||
dashboard=False,
|
||||
streaming_on=True,
|
||||
verbose=True,
|
||||
stopping_token="<DONE>",
|
||||
state_save_file_type="json",
|
||||
saved_state_path="processor_agent.json",
|
||||
)
|
||||
|
||||
# Initialize Claims Auditor agent
|
||||
auditor_agent = Agent(
|
||||
agent_name="ClaimsAuditor",
|
||||
agent_description="Audits processed claims for accuracy and compliance with policies and regulations",
|
||||
system_prompt="""Audit processed insurance claims for accuracy and compliance. Review claim decisions,
|
||||
identify potential fraud or errors, and ensure all processing follows established guidelines and regulations.""",
|
||||
llm=model,
|
||||
max_loops=1,
|
||||
dashboard=False,
|
||||
streaming_on=True,
|
||||
verbose=True,
|
||||
stopping_token="<DONE>",
|
||||
state_save_file_type="json",
|
||||
saved_state_path="auditor_agent.json",
|
||||
)
|
||||
|
||||
# Create a list of agents
|
||||
agents = [director_agent, processor_agent, auditor_agent]
|
||||
|
||||
print(showcase_available_agents(agents=agents))
|
||||
@ -0,0 +1,114 @@
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Swarm imports
|
||||
from swarms.structs.agent import Agent
|
||||
from swarms.structs.hiearchical_swarm import (
|
||||
HierarchicalSwarm,
|
||||
SwarmSpec,
|
||||
OrganizationalUnit,
|
||||
)
|
||||
from swarms.utils.function_caller_model import OpenAIFunctionCaller
|
||||
|
||||
# Load environment variables
|
||||
load_dotenv()
|
||||
|
||||
# Create the agents first
|
||||
research_manager = Agent(
|
||||
agent_name="Research Manager",
|
||||
agent_description="Manages research operations and coordinates research tasks",
|
||||
system_prompt="You are a research manager responsible for overseeing research projects and coordinating research efforts.",
|
||||
model_name="gpt-4o",
|
||||
)
|
||||
|
||||
data_analyst = Agent(
|
||||
agent_name="Data Analyst",
|
||||
agent_description="Analyzes data and generates insights",
|
||||
system_prompt="You are a data analyst specializing in processing and analyzing data to extract meaningful insights.",
|
||||
model_name="gpt-4o",
|
||||
)
|
||||
|
||||
research_assistant = Agent(
|
||||
agent_name="Research Assistant",
|
||||
agent_description="Assists with research tasks and data collection",
|
||||
system_prompt="You are a research assistant who helps gather information and support research activities.",
|
||||
model_name="gpt-4o",
|
||||
)
|
||||
|
||||
development_manager = Agent(
|
||||
agent_name="Development Manager",
|
||||
agent_description="Manages development projects and coordinates development tasks",
|
||||
system_prompt="You are a development manager responsible for overseeing software development projects and coordinating development efforts.",
|
||||
model_name="gpt-4o",
|
||||
)
|
||||
|
||||
software_engineer = Agent(
|
||||
agent_name="Software Engineer",
|
||||
agent_description="Develops and implements software solutions",
|
||||
system_prompt="You are a software engineer specializing in building and implementing software solutions.",
|
||||
model_name="gpt-4o",
|
||||
)
|
||||
|
||||
qa_engineer = Agent(
|
||||
agent_name="QA Engineer",
|
||||
agent_description="Tests and ensures quality of software",
|
||||
system_prompt="You are a QA engineer responsible for testing software and ensuring its quality.",
|
||||
model_name="gpt-4o",
|
||||
)
|
||||
|
||||
# Create organizational units with the agents
|
||||
research_unit = OrganizationalUnit(
|
||||
name="Research Unit",
|
||||
description="Handles research and analysis tasks",
|
||||
manager=research_manager,
|
||||
members=[data_analyst, research_assistant],
|
||||
)
|
||||
|
||||
development_unit = OrganizationalUnit(
|
||||
name="Development Unit",
|
||||
description="Handles development and implementation tasks",
|
||||
manager=development_manager,
|
||||
members=[software_engineer, qa_engineer],
|
||||
)
|
||||
|
||||
# Initialize the director agent
|
||||
director = OpenAIFunctionCaller(
|
||||
model_name="gpt-4o",
|
||||
system_prompt=(
|
||||
"As the Director of this Hierarchical Agent Swarm, you are responsible for:\n"
|
||||
"1. Analyzing tasks and breaking them down into subtasks\n"
|
||||
"2. Assigning tasks to appropriate organizational units\n"
|
||||
"3. Coordinating communication between units\n"
|
||||
"4. Ensuring tasks are completed efficiently and effectively\n"
|
||||
"5. Providing feedback and guidance to units as needed\n\n"
|
||||
"Your decisions should be based on the capabilities of each unit and the requirements of the task."
|
||||
),
|
||||
api_key=os.getenv("OPENAI_API_KEY"),
|
||||
temperature=0.5,
|
||||
base_model=SwarmSpec,
|
||||
max_tokens=10000,
|
||||
)
|
||||
|
||||
# Initialize the hierarchical swarm with the organizational units
|
||||
swarm = HierarchicalSwarm(
|
||||
name="Example Hierarchical Swarm",
|
||||
description="A hierarchical swarm demonstrating multi-unit collaboration",
|
||||
director=director,
|
||||
organizational_units=[research_unit, development_unit],
|
||||
max_loops=2, # Allow for feedback and iteration
|
||||
output_type="dict",
|
||||
)
|
||||
|
||||
# Example task to run through the swarm
|
||||
task = """
|
||||
Develop a comprehensive market analysis for a new AI-powered productivity tool.
|
||||
The analysis should include:
|
||||
1. Market research and competitor analysis
|
||||
2. User needs and pain points
|
||||
3. Technical feasibility assessment
|
||||
4. Implementation recommendations
|
||||
"""
|
||||
|
||||
# Run the task through the swarm
|
||||
result = swarm.run(task)
|
||||
print("Swarm Results:", result)
|
||||
@ -0,0 +1,411 @@
|
||||
from functools import lru_cache
|
||||
from io import BytesIO
|
||||
import os
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from typing import List, Optional, Union
|
||||
|
||||
import PyPDF2
|
||||
|
||||
from swarms.structs.agent import Agent
|
||||
from swarms.structs.conversation import Conversation
|
||||
from swarms.structs.ma_utils import set_random_models_for_agents
|
||||
from swarms.utils.history_output_formatter import (
|
||||
history_output_formatter,
|
||||
)
|
||||
|
||||
from swarms.utils.generate_keys import generate_api_key
|
||||
|
||||
# System prompts for each agent
|
||||
CLARA_SYS_PROMPT = """
|
||||
You are Clara, a meticulous and client-focused Criteria Agent specialized in understanding and validating contract requirements for a legal automation system. Your purpose is to ensure all contract criteria are clear, complete, and actionable.
|
||||
|
||||
KEY RESPONSIBILITIES:
|
||||
- Extract and interpret contract requirements from client documents, text, or instructions.
|
||||
- Validate criteria for completeness, consistency, and legal feasibility.
|
||||
- Identify ambiguities, missing details, or potential risks in the requirements.
|
||||
- Produce a clear, structured summary of the criteria for downstream use.
|
||||
|
||||
APPROACH:
|
||||
- Begin with a professional introduction explaining your role in ensuring contract clarity.
|
||||
- Ask targeted questions to resolve ambiguities or fill gaps in the criteria.
|
||||
- Summarize and confirm requirements to ensure accuracy.
|
||||
- Flag any criteria that may lead to legal or practical issues.
|
||||
- Maintain strict confidentiality and data security.
|
||||
|
||||
OUTPUT FORMAT:
|
||||
Provide a plain text summary with:
|
||||
1. Validated contract criteria (e.g., parties, purpose, terms, jurisdiction).
|
||||
2. Notes on any ambiguities or missing information.
|
||||
3. Recommendations for clarifying or refining criteria.
|
||||
"""
|
||||
|
||||
MASON_SYS_PROMPT = """
|
||||
You are Mason, a precise and creative Contract Drafting Agent specialized in crafting exceptional, extensive legal contracts for a legal automation system. Your expertise lies in producing long, comprehensive, enforceable, and tailored contract documents that cover all possible contingencies and details.
|
||||
|
||||
KEY RESPONSIBILITIES:
|
||||
- Draft detailed, lengthy contracts based on validated criteria provided.
|
||||
- Ensure contracts are legally sound, exhaustive, and client-specific, addressing all relevant aspects thoroughly.
|
||||
- Use precise language while maintaining accessibility for non-legal readers.
|
||||
- Incorporate feedback from evaluations to refine and enhance drafts.
|
||||
- Include extensive clauses to cover all potential scenarios, risks, and obligations.
|
||||
|
||||
APPROACH:
|
||||
- Structure contracts with clear, detailed sections and consistent formatting.
|
||||
- Include all essential elements (parties, purpose, terms, signatures, etc.) with comprehensive elaboration.
|
||||
- Tailor clauses to address specific client needs, jurisdictional requirements, and potential future disputes.
|
||||
- Provide in-depth explanations of terms, conditions, and contingencies.
|
||||
- Highlight areas requiring further review or customization.
|
||||
- Output the contract as a plain text string, avoiding markdown.
|
||||
|
||||
OUTPUT FORMAT:
|
||||
Provide a plain text contract with:
|
||||
1. Identification of parties and effective date.
|
||||
2. Detailed statement of purpose and scope.
|
||||
3. Exhaustive terms and conditions covering all possible scenarios.
|
||||
4. Comprehensive rights and obligations of each party.
|
||||
5. Detailed termination and amendment procedures.
|
||||
6. Signature blocks.
|
||||
7. Annotations for areas needing review (in comments).
|
||||
"""
|
||||
|
||||
SOPHIA_SYS_PROMPT = """
|
||||
You are Sophia, a rigorous and insightful Contract Evaluation Agent specialized in reviewing and improving legal contracts for a legal automation system. Your role is to evaluate contracts for quality, compliance, and clarity, providing actionable feedback to enhance the final document.
|
||||
|
||||
KEY RESPONSIBILITIES:
|
||||
- Review draft contracts for legal risks, clarity, and enforceability.
|
||||
- Identify compliance issues with relevant laws and regulations.
|
||||
- Assess whether the contract meets the provided criteria and client needs.
|
||||
- Provide specific, actionable feedback for revisions.
|
||||
- Recommend areas requiring human attorney review.
|
||||
|
||||
APPROACH:
|
||||
- Begin with a disclaimer that your evaluation is automated and not a substitute for human legal advice.
|
||||
- Analyze the contract section by section, focusing on legal soundness and clarity.
|
||||
- Highlight strengths and weaknesses, with emphasis on areas for improvement.
|
||||
- Provide precise suggestions for revised language or additional clauses.
|
||||
- Maintain a professional, constructive tone to support iterative improvement.
|
||||
|
||||
OUTPUT FORMAT:
|
||||
Provide a plain text evaluation with:
|
||||
1. Summary of the contract's strengths.
|
||||
2. Identified issues (legal risks, ambiguities, missing elements).
|
||||
3. Specific feedback for revisions (e.g., suggested clause changes).
|
||||
4. Compliance notes (e.g., relevant laws or regulations).
|
||||
5. Recommendations for human attorney review.
|
||||
"""
|
||||
|
||||
|
||||
class LegalSwarm:
|
||||
def __init__(
|
||||
self,
|
||||
name: str = "Legal Swarm",
|
||||
description: str = "A swarm of agents that can handle legal tasks",
|
||||
max_loops: int = 1,
|
||||
user_name: str = "John Doe",
|
||||
documents: Optional[List[str]] = None,
|
||||
output_type: str = "list",
|
||||
):
|
||||
"""
|
||||
Initialize the LegalSwarm with a base LLM and configure agents.
|
||||
|
||||
Args:
|
||||
llm (BaseLLM): The underlying LLM model for all agents.
|
||||
max_loops (int): Maximum iterations for each agent's task.
|
||||
"""
|
||||
self.max_loops = max_loops
|
||||
self.name = name
|
||||
self.description = description
|
||||
self.user_name = user_name
|
||||
self.documents = documents
|
||||
self.output_type = output_type
|
||||
|
||||
self.agents = self._initialize_agents()
|
||||
self.agents = set_random_models_for_agents(self.agents)
|
||||
self.conversation = Conversation()
|
||||
self.handle_initial_processing()
|
||||
|
||||
def handle_initial_processing(self):
|
||||
if self.documents:
|
||||
documents_text = self.handle_documents(self.documents)
|
||||
else:
|
||||
documents_text = None
|
||||
|
||||
self.conversation.add(
|
||||
role=self.user_name,
|
||||
content=f"Firm Name: {self.name}\nFirm Description: {self.description}\nUser Name: {self.user_name}\nDocuments: {documents_text}",
|
||||
)
|
||||
|
||||
def _initialize_agents(self) -> List[Agent]:
|
||||
"""
|
||||
Initialize all agents with their respective prompts and configurations.
|
||||
|
||||
Returns:
|
||||
List[Agent]: List of Agent instances.
|
||||
"""
|
||||
return [
|
||||
Agent(
|
||||
agent_name="Clara-Intake-Agent",
|
||||
agent_description="Handles client data intake and validation",
|
||||
system_prompt=CLARA_SYS_PROMPT,
|
||||
max_loops=self.max_loops,
|
||||
dynamic_temperature_enabled=True,
|
||||
output_type="final",
|
||||
),
|
||||
# Agent(
|
||||
# agent_name="Riley-Report-Agent",
|
||||
# agent_description="Generates client reports from intake data",
|
||||
# system_prompt=RILEY_SYS_PROMPT,
|
||||
# max_loops=self.max_loops,
|
||||
# dynamic_temperature_enabled=True,
|
||||
# output_type = "final"
|
||||
# ),
|
||||
Agent(
|
||||
agent_name="Mason-Contract-Agent",
|
||||
agent_description="Creates and updates legal contracts",
|
||||
system_prompt=MASON_SYS_PROMPT,
|
||||
max_loops=self.max_loops,
|
||||
dynamic_temperature_enabled=True,
|
||||
output_type="final",
|
||||
),
|
||||
Agent(
|
||||
agent_name="Sophia-Counsel-Agent",
|
||||
agent_description="Provides legal advice and compliance checks",
|
||||
system_prompt=SOPHIA_SYS_PROMPT,
|
||||
max_loops=self.max_loops,
|
||||
dynamic_temperature_enabled=True,
|
||||
output_type="final",
|
||||
),
|
||||
# Agent(
|
||||
# agent_name="Ethan-Coordinator-Agent",
|
||||
# agent_description="Manages workflow and communication",
|
||||
# system_prompt=ETHAN_SYS_PROMPT,
|
||||
# max_loops=self.max_loops,
|
||||
# dynamic_temperature_enabled=True,
|
||||
# output_type = "final"
|
||||
# ),
|
||||
]
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def handle_documents(self, documents: List[str]) -> str:
|
||||
"""
|
||||
Handle a list of documents concurrently, extracting text from PDFs and other documents.
|
||||
|
||||
Args:
|
||||
documents (List[str]): List of document file paths to process.
|
||||
|
||||
Returns:
|
||||
str: Combined text content from all documents.
|
||||
"""
|
||||
|
||||
def process_document(file_path: str) -> str:
|
||||
"""Process a single document and return its text content."""
|
||||
if not os.path.exists(file_path):
|
||||
return f"Error: File not found - {file_path}"
|
||||
|
||||
try:
|
||||
if file_path.lower().endswith(".pdf"):
|
||||
with open(file_path, "rb") as file:
|
||||
pdf_reader = PyPDF2.PdfReader(file)
|
||||
text = ""
|
||||
for page in pdf_reader.pages:
|
||||
text += page.extract_text() + "\n"
|
||||
return text
|
||||
else:
|
||||
# Handle other document types (txt, docx, etc.)
|
||||
with open(
|
||||
file_path, "r", encoding="utf-8"
|
||||
) as file:
|
||||
return file.read()
|
||||
except Exception as e:
|
||||
return f"Error processing {file_path}: {str(e)}"
|
||||
|
||||
# Process documents concurrently
|
||||
combined_text = ""
|
||||
with ThreadPoolExecutor(
|
||||
max_workers=min(len(documents), 4)
|
||||
) as executor:
|
||||
results = list(executor.map(process_document, documents))
|
||||
combined_text = "\n\n".join(results)
|
||||
|
||||
return combined_text
|
||||
|
||||
def find_agent_by_name(self, name: str) -> Agent:
|
||||
"""
|
||||
Find an agent by their name.
|
||||
"""
|
||||
for agent in self.agents:
|
||||
if agent.agent_name == name:
|
||||
return agent
|
||||
|
||||
def initial_processing(self):
|
||||
clara_agent = self.find_agent_by_name("Clara-Intake-Agent")
|
||||
|
||||
# Run Clara's agent
|
||||
clara_output = clara_agent.run(
|
||||
f"History: {self.conversation.get_str()}\n Create a structured summary document of the customer's case."
|
||||
)
|
||||
|
||||
self.conversation.add(
|
||||
role="Clara-Intake-Agent", content=clara_output
|
||||
)
|
||||
|
||||
def create_contract(self, task: str):
|
||||
mason_agent = self.find_agent_by_name("Mason-Contract-Agent")
|
||||
|
||||
mason_output = mason_agent.run(
|
||||
f"History: {self.conversation.get_str()}\n Your purpose is to create a contract based on the following details: {task}"
|
||||
)
|
||||
|
||||
self.conversation.add(
|
||||
role="Mason-Contract-Agent", content=mason_output
|
||||
)
|
||||
|
||||
artifact_id = generate_api_key(
|
||||
"legal-swarm-artifact-", length=10
|
||||
)
|
||||
|
||||
# Run Sophia's agent
|
||||
sophia_agent = self.find_agent_by_name("Sophia-Counsel-Agent")
|
||||
sophia_output = sophia_agent.run(
|
||||
f"History: {self.conversation.get_str()}\n Your purpose is to review the contract Mason created and provide feedback."
|
||||
)
|
||||
|
||||
self.conversation.add(
|
||||
role="Sophia-Counsel-Agent", content=sophia_output
|
||||
)
|
||||
|
||||
# Run Mason's agent
|
||||
mason_agent = self.find_agent_by_name("Mason-Contract-Agent")
|
||||
mason_output = mason_agent.run(
|
||||
f"History: {self.conversation.get_str()}\n Your purpose is to update the contract based on the feedback Sophia provided."
|
||||
)
|
||||
|
||||
self.conversation.add(
|
||||
role="Mason-Contract-Agent", content=mason_output
|
||||
)
|
||||
|
||||
self.create_pdf_from_string(
|
||||
mason_output, f"{artifact_id}-contract.pdf"
|
||||
)
|
||||
|
||||
def create_pdf_from_string(
|
||||
self, string: str, output_path: Optional[str] = None
|
||||
) -> Union[BytesIO, str]:
|
||||
"""
|
||||
Create a PDF from a string with proper formatting and styling.
|
||||
|
||||
Args:
|
||||
string (str): The text content to convert to PDF
|
||||
output_path (Optional[str]): If provided, save the PDF to this path. Otherwise return BytesIO object
|
||||
|
||||
Returns:
|
||||
Union[BytesIO, str]: Either a BytesIO object containing the PDF or the path where it was saved
|
||||
"""
|
||||
try:
|
||||
from reportlab.lib.pagesizes import letter
|
||||
from reportlab.pdfgen import canvas
|
||||
from reportlab.lib.styles import (
|
||||
getSampleStyleSheet,
|
||||
ParagraphStyle,
|
||||
)
|
||||
from reportlab.platypus import (
|
||||
Paragraph,
|
||||
SimpleDocTemplate,
|
||||
)
|
||||
from reportlab.lib.units import inch
|
||||
|
||||
# Create a buffer or file
|
||||
if output_path:
|
||||
doc = SimpleDocTemplate(output_path, pagesize=letter)
|
||||
else:
|
||||
buffer = BytesIO()
|
||||
doc = SimpleDocTemplate(buffer, pagesize=letter)
|
||||
|
||||
# Create styles
|
||||
styles = getSampleStyleSheet()
|
||||
custom_style = ParagraphStyle(
|
||||
"CustomStyle",
|
||||
parent=styles["Normal"],
|
||||
fontSize=12,
|
||||
leading=14,
|
||||
spaceAfter=12,
|
||||
firstLineIndent=0.5 * inch,
|
||||
)
|
||||
|
||||
# Prepare content
|
||||
story = []
|
||||
paragraphs = string.split("\n\n")
|
||||
|
||||
for para in paragraphs:
|
||||
if para.strip():
|
||||
story.append(
|
||||
Paragraph(para.strip(), custom_style)
|
||||
)
|
||||
|
||||
# Build PDF
|
||||
doc.build(story)
|
||||
|
||||
if output_path:
|
||||
return output_path
|
||||
else:
|
||||
buffer.seek(0)
|
||||
return buffer
|
||||
|
||||
except ImportError:
|
||||
raise ImportError(
|
||||
"Please install reportlab: pip install reportlab"
|
||||
)
|
||||
except Exception as e:
|
||||
raise Exception(f"Error creating PDF: {str(e)}")
|
||||
|
||||
def run(self, task: str):
|
||||
"""
|
||||
Process an input document through the swarm, coordinating tasks among agents.
|
||||
|
||||
Args:
|
||||
task (str): The input task or text to process.
|
||||
|
||||
Returns:
|
||||
Dict[str, Any]: Final output including client data, report, contract, counsel, and workflow status.
|
||||
"""
|
||||
self.conversation.add(role=self.user_name, content=task)
|
||||
|
||||
self.initial_processing()
|
||||
|
||||
self.create_contract(task)
|
||||
|
||||
return history_output_formatter(
|
||||
self.conversation, type=self.output_type
|
||||
)
|
||||
|
||||
|
||||
# Example usage
|
||||
if __name__ == "__main__":
|
||||
# Initialize the swarm
|
||||
swarm = LegalSwarm(
|
||||
max_loops=1,
|
||||
name="TGSC's Legal Swarm",
|
||||
description="A swarm of agents that can handle legal tasks",
|
||||
user_name="Kye Gomez",
|
||||
output_type="json",
|
||||
)
|
||||
|
||||
# Sample document for COO employment contract
|
||||
sample_document = """
|
||||
Company: Swarms TGSC
|
||||
Entity Type: Delaware C Corporation
|
||||
Position: Chief Operating Officer (COO)
|
||||
Details: Creating an employment contract for a COO position with standard executive-level terms including:
|
||||
- Base salary and equity compensation $5,000,000
|
||||
- Performance bonuses and incentives
|
||||
- Benefits package
|
||||
- Non-compete and confidentiality clauses
|
||||
- Termination provisions
|
||||
- Stock options and vesting schedule
|
||||
- Reporting structure and responsibilities
|
||||
Contact: hr@swarms.tgsc
|
||||
"""
|
||||
|
||||
# Run the swarm
|
||||
result = swarm.run(task=sample_document)
|
||||
print("Swarm Output:", result)
|
||||
@ -1,15 +1,4 @@
|
||||
from dotenv import load_dotenv
|
||||
|
||||
from swarms import Agent
|
||||
from swarms.tools.mcp_integration import MCPServerSseParams
|
||||
|
||||
load_dotenv()
|
||||
|
||||
|
||||
server = MCPServerSseParams(
|
||||
url="http://localhost:8000/sse",
|
||||
timeout=10,
|
||||
)
|
||||
|
||||
tools = [
|
||||
{
|
||||
@ -0,0 +1,20 @@
|
||||
from swarms import ReasoningAgentRouter
|
||||
|
||||
|
||||
calculus_router = ReasoningAgentRouter(
|
||||
agent_name="calculus-expert",
|
||||
description="A calculus problem solving agent",
|
||||
model_name="gpt-4o-mini",
|
||||
system_prompt="You are a calculus expert. Solve differentiation and integration problems methodically.",
|
||||
swarm_type="self-consistency",
|
||||
num_samples=3, # Generate 3 samples to ensure consistency
|
||||
output_type="list",
|
||||
)
|
||||
|
||||
|
||||
# Example calculus problem
|
||||
calculus_problem = "Find the derivative of f(x) = x³ln(x) - 5x²"
|
||||
|
||||
# Get the solution
|
||||
solution = calculus_router.run(calculus_problem)
|
||||
print(solution)
|
||||
@ -0,0 +1,69 @@
|
||||
from swarms.structs.agent import Agent
|
||||
from swarms.structs.swarm_router import SwarmRouter
|
||||
import json
|
||||
|
||||
# Create the agents first
|
||||
research_manager = Agent(
|
||||
agent_name="Research Manager",
|
||||
agent_description="Manages research operations and coordinates research tasks",
|
||||
system_prompt="You are a research manager responsible for overseeing research projects and coordinating research efforts.",
|
||||
model_name="gpt-4o",
|
||||
)
|
||||
|
||||
data_analyst = Agent(
|
||||
agent_name="Data Analyst",
|
||||
agent_description="Analyzes data and generates insights",
|
||||
system_prompt="You are a data analyst specializing in processing and analyzing data to extract meaningful insights.",
|
||||
model_name="gpt-4o",
|
||||
)
|
||||
|
||||
research_assistant = Agent(
|
||||
agent_name="Research Assistant",
|
||||
agent_description="Assists with research tasks and data collection",
|
||||
system_prompt="You are a research assistant who helps gather information and support research activities.",
|
||||
model_name="gpt-4o",
|
||||
)
|
||||
|
||||
development_manager = Agent(
|
||||
agent_name="Development Manager",
|
||||
agent_description="Manages development projects and coordinates development tasks",
|
||||
system_prompt="You are a development manager responsible for overseeing software development projects and coordinating development efforts.",
|
||||
model_name="gpt-4o",
|
||||
)
|
||||
|
||||
software_engineer = Agent(
|
||||
agent_name="Software Engineer",
|
||||
agent_description="Develops and implements software solutions",
|
||||
system_prompt="You are a software engineer specializing in building and implementing software solutions.",
|
||||
model_name="gpt-4o",
|
||||
)
|
||||
|
||||
qa_engineer = Agent(
|
||||
agent_name="QA Engineer",
|
||||
agent_description="Tests and ensures quality of software",
|
||||
system_prompt="You are a QA engineer responsible for testing software and ensuring its quality.",
|
||||
model_name="gpt-4o",
|
||||
)
|
||||
|
||||
swarm_router = SwarmRouter(
|
||||
name="Swarm Router",
|
||||
description="A swarm router that routes tasks to the appropriate agents",
|
||||
agents=[
|
||||
research_manager,
|
||||
data_analyst,
|
||||
research_assistant,
|
||||
development_manager,
|
||||
software_engineer,
|
||||
qa_engineer,
|
||||
],
|
||||
multi_agent_collab_prompt=True,
|
||||
swarm_type="MixtureOfAgents",
|
||||
output_type="dict",
|
||||
)
|
||||
|
||||
output = swarm_router.run(
|
||||
task="Write a research paper on the impact of AI on the future of work"
|
||||
)
|
||||
|
||||
with open("output.json", "w") as f:
|
||||
json.dump(output, f)
|
||||
@ -1,87 +0,0 @@
|
||||
from swarms.structs.agent import Agent
|
||||
from typing import List
|
||||
|
||||
|
||||
def showcase_available_agents(
|
||||
agents: List[Agent],
|
||||
name: str = None,
|
||||
description: str = None,
|
||||
format: str = "XML",
|
||||
) -> str:
|
||||
"""
|
||||
Format the available agents in either XML or Table format.
|
||||
|
||||
Args:
|
||||
agents (List[Agent]): A list of agents to represent
|
||||
name (str, optional): Name of the swarm
|
||||
description (str, optional): Description of the swarm
|
||||
format (str, optional): Output format ("XML" or "Table"). Defaults to "XML"
|
||||
|
||||
Returns:
|
||||
str: Formatted string containing agent information
|
||||
"""
|
||||
|
||||
def truncate(text: str, max_length: int = 130) -> str:
|
||||
return (
|
||||
f"{text[:max_length]}..."
|
||||
if len(text) > max_length
|
||||
else text
|
||||
)
|
||||
|
||||
output = []
|
||||
|
||||
if format.upper() == "TABLE":
|
||||
output.append("\n| ID | Agent Name | Description |")
|
||||
output.append("|-----|------------|-------------|")
|
||||
for idx, agent in enumerate(agents):
|
||||
if isinstance(agent, Agent):
|
||||
agent_name = getattr(agent, "agent_name", str(agent))
|
||||
description = getattr(
|
||||
agent,
|
||||
"description",
|
||||
getattr(
|
||||
agent, "system_prompt", "Unknown description"
|
||||
),
|
||||
)
|
||||
desc = truncate(description, 50)
|
||||
output.append(
|
||||
f"| {idx + 1} | {agent_name} | {desc} |"
|
||||
)
|
||||
else:
|
||||
output.append(
|
||||
f"| {idx + 1} | {agent} | Unknown description |"
|
||||
)
|
||||
return "\n".join(output)
|
||||
|
||||
# Default XML format
|
||||
output.append("<agents>")
|
||||
if name:
|
||||
output.append(f" <name>{name}</name>")
|
||||
if description:
|
||||
output.append(
|
||||
f" <description>{truncate(description)}</description>"
|
||||
)
|
||||
for idx, agent in enumerate(agents):
|
||||
output.append(f" <agent id='{idx + 1}'>")
|
||||
if isinstance(agent, Agent):
|
||||
agent_name = getattr(agent, "agent_name", str(agent))
|
||||
description = getattr(
|
||||
agent,
|
||||
"description",
|
||||
getattr(
|
||||
agent, "system_prompt", "Unknown description"
|
||||
),
|
||||
)
|
||||
output.append(f" <name>{agent_name}</name>")
|
||||
output.append(
|
||||
f" <description>{truncate(description)}</description>"
|
||||
)
|
||||
else:
|
||||
output.append(f" <name>{agent}</name>")
|
||||
output.append(
|
||||
" <description>Unknown description</description>"
|
||||
)
|
||||
output.append(" </agent>")
|
||||
output.append("</agents>")
|
||||
|
||||
return "\n".join(output)
|
||||
@ -0,0 +1,96 @@
|
||||
from swarms.structs.agent import Agent
|
||||
from typing import List, Any, Optional, Union
|
||||
import random
|
||||
|
||||
|
||||
def list_all_agents(
|
||||
agents: List[Union[Agent, Any]],
|
||||
conversation: Optional[Any] = None,
|
||||
name: str = "",
|
||||
add_to_conversation: bool = False,
|
||||
) -> str:
|
||||
"""Lists all agents in a swarm and optionally adds them to a conversation.
|
||||
|
||||
This function compiles information about all agents in a swarm, including their names and descriptions.
|
||||
It can optionally add this information to a conversation history.
|
||||
|
||||
Args:
|
||||
agents (List[Union[Agent, Any]]): List of agents to list information about
|
||||
conversation (Any): Conversation object to optionally add agent information to
|
||||
name (str): Name of the swarm/group of agents
|
||||
add_to_conversation (bool, optional): Whether to add agent information to conversation. Defaults to False.
|
||||
|
||||
Returns:
|
||||
str: Formatted string containing information about all agents
|
||||
|
||||
Example:
|
||||
>>> agents = [agent1, agent2]
|
||||
>>> conversation = Conversation()
|
||||
>>> agent_info = list_all_agents(agents, conversation, "MySwarm")
|
||||
>>> print(agent_info)
|
||||
Total Agents: 2
|
||||
|
||||
Agent: Agent1
|
||||
Description: First agent description...
|
||||
|
||||
Agent: Agent2
|
||||
Description: Second agent description...
|
||||
"""
|
||||
|
||||
# Compile information about all agents
|
||||
total_agents = len(agents)
|
||||
|
||||
all_agents = f"Total Agents: {total_agents}\n\n" + "\n\n".join(
|
||||
f"Agent: {agent.agent_name} \n\n Description: {agent.description or (agent.system_prompt[:50] + '...' if len(agent.system_prompt) > 50 else agent.system_prompt)}"
|
||||
for agent in agents
|
||||
)
|
||||
|
||||
if add_to_conversation:
|
||||
# Add the agent information to the conversation
|
||||
conversation.add(
|
||||
role="System",
|
||||
content=f"All Agents Available in the Swarm {name}:\n\n{all_agents}",
|
||||
)
|
||||
|
||||
return all_agents
|
||||
|
||||
|
||||
models = [
|
||||
"anthropic/claude-3-sonnet-20240229",
|
||||
"openai/gpt-4o-mini",
|
||||
"openai/gpt-4o",
|
||||
"deepseek/deepseek-chat",
|
||||
"deepseek/deepseek-reasoner",
|
||||
"groq/deepseek-r1-distill-qwen-32b",
|
||||
"groq/deepseek-r1-distill-qwen-32b",
|
||||
# "gemini/gemini-pro",
|
||||
# "gemini/gemini-1.5-pro",
|
||||
"openai/03-mini",
|
||||
"o4-mini",
|
||||
"o3",
|
||||
"gpt-4.1",
|
||||
"gpt-4.1-nano",
|
||||
]
|
||||
|
||||
|
||||
def set_random_models_for_agents(
|
||||
agents: Union[List[Agent], Agent], model_names: List[str] = models
|
||||
) -> Union[List[Agent], Agent]:
|
||||
"""Sets random models for agents in the swarm.
|
||||
|
||||
Args:
|
||||
agents (Union[List[Agent], Agent]): Either a single agent or a list of agents
|
||||
model_names (List[str], optional): List of model names to choose from. Defaults to models.
|
||||
|
||||
Returns:
|
||||
Union[List[Agent], Agent]: The agent(s) with randomly assigned models
|
||||
"""
|
||||
if isinstance(agents, list):
|
||||
return [
|
||||
setattr(agent, "model_name", random.choice(model_names))
|
||||
or agent
|
||||
for agent in agents
|
||||
]
|
||||
else:
|
||||
setattr(agents, "model_name", random.choice(model_names))
|
||||
return agents
|
||||
@ -1,19 +1,6 @@
|
||||
from typing import Literal
|
||||
|
||||
# Literal of output types
|
||||
OutputType = Literal[
|
||||
"all",
|
||||
"final",
|
||||
"list",
|
||||
"dict",
|
||||
".json",
|
||||
".md",
|
||||
".txt",
|
||||
".yaml",
|
||||
".toml",
|
||||
"string",
|
||||
"str",
|
||||
]
|
||||
from swarms.utils.history_output_formatter import (
|
||||
HistoryOutputType as OutputType,
|
||||
)
|
||||
|
||||
# Use the OutputType for type annotations
|
||||
output_type: OutputType
|
||||
|
||||
@ -1,395 +0,0 @@
|
||||
import os
|
||||
from typing import List, Optional
|
||||
|
||||
from loguru import logger
|
||||
from pydantic import BaseModel, Field
|
||||
from pydantic.v1 import validator
|
||||
from tenacity import (
|
||||
retry,
|
||||
stop_after_attempt,
|
||||
wait_exponential,
|
||||
)
|
||||
|
||||
from swarms.structs.agent import Agent
|
||||
from swarms.structs.swarm_router import SwarmRouter, SwarmType
|
||||
from swarms.utils.function_caller_model import OpenAIFunctionCaller
|
||||
|
||||
|
||||
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 AgentConfig(BaseModel):
|
||||
"""Configuration for an individual agent in a swarm"""
|
||||
|
||||
name: str = Field(
|
||||
description="The name of the agent",
|
||||
)
|
||||
description: str = Field(
|
||||
description="A description of the agent's purpose and capabilities",
|
||||
)
|
||||
system_prompt: str = Field(
|
||||
description="The system prompt that defines the agent's behavior",
|
||||
)
|
||||
|
||||
|
||||
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",
|
||||
)
|
||||
max_loops: int = Field(
|
||||
description="The maximum number of loops for the swarm to iterate on",
|
||||
)
|
||||
|
||||
@validator("agents")
|
||||
def validate_agents(cls, v):
|
||||
if not v:
|
||||
raise ValueError("Swarm must have at least one agent")
|
||||
return v
|
||||
|
||||
|
||||
class AutoSwarmBuilderOutput(BaseModel):
|
||||
"""A class that automatically builds and manages swarms of AI agents with enhanced error handling."""
|
||||
|
||||
name: Optional[str] = Field(
|
||||
description="The name of the swarm",
|
||||
example="DefaultSwarm",
|
||||
default=None,
|
||||
)
|
||||
description: Optional[str] = Field(
|
||||
description="The description of the swarm's purpose and capabilities",
|
||||
example="Generic AI Agent Swarm",
|
||||
default=None,
|
||||
)
|
||||
verbose: Optional[bool] = Field(
|
||||
description="Whether to display verbose output",
|
||||
default=None,
|
||||
)
|
||||
model_name: Optional[str] = Field(
|
||||
description="The name of the OpenAI model to use",
|
||||
default=None,
|
||||
)
|
||||
boss_output_schema: Optional[list] = Field(
|
||||
description="The schema for the output of the BOSS system prompt",
|
||||
default=None,
|
||||
)
|
||||
director_agents_created: Optional[SwarmConfig] = Field(
|
||||
description="The agents created by the director",
|
||||
default=None,
|
||||
)
|
||||
swarm_router_outputs: Optional[list] = Field(
|
||||
description="The outputs from the swarm router",
|
||||
default=None,
|
||||
)
|
||||
max_loops: Optional[int] = Field(
|
||||
description="The maximum number of loops for the swarm to iterate on",
|
||||
default=None,
|
||||
)
|
||||
swarm_type: Optional[SwarmType] = Field(
|
||||
description="The type of swarm to build",
|
||||
default=None,
|
||||
)
|
||||
|
||||
|
||||
class AutoSwarmBuilder:
|
||||
"""A class that automatically builds and manages swarms of AI agents with enhanced error handling."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
name: Optional[str] = "autonomous-swarm-builder",
|
||||
description: Optional[
|
||||
str
|
||||
] = "Given a task, this swarm will automatically create specialized agents and route it to the appropriate agents.",
|
||||
verbose: bool = True,
|
||||
model_name: str = "gpt-4o",
|
||||
boss_output_schema: list = None,
|
||||
swarm_router_outputs: AutoSwarmBuilderOutput = None,
|
||||
max_loops: int = 1,
|
||||
swarm_type: str = "SequentialWorkflow",
|
||||
auto_generate_prompts_for_agents: bool = False,
|
||||
shared_memory_system: callable = None,
|
||||
):
|
||||
self.name = name or "DefaultSwarm"
|
||||
self.description = description or "Generic AI Agent Swarm"
|
||||
self.verbose = verbose
|
||||
self.agents_pool = []
|
||||
self.api_key = os.getenv("OPENAI_API_KEY")
|
||||
self.model_name = model_name
|
||||
self.boss_output_schema = boss_output_schema
|
||||
self.max_loops = max_loops
|
||||
self.swarm_type = swarm_type
|
||||
self.auto_generate_prompts_for_agents = (
|
||||
auto_generate_prompts_for_agents
|
||||
)
|
||||
self.shared_memory_system = shared_memory_system
|
||||
self.auto_swarm_builder_output = AutoSwarmBuilderOutput(
|
||||
name=name,
|
||||
description=description,
|
||||
verbose=verbose,
|
||||
model_name=model_name,
|
||||
boss_output_schema=boss_output_schema or [],
|
||||
swarm_router_outputs=swarm_router_outputs or [],
|
||||
max_loops=max_loops,
|
||||
swarm_type=swarm_type,
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"Initialized AutoSwarmBuilder",
|
||||
extra={
|
||||
"swarm_name": self.name,
|
||||
"description": self.description,
|
||||
"model": self.model_name,
|
||||
},
|
||||
)
|
||||
|
||||
def run(
|
||||
self,
|
||||
task: str,
|
||||
image_url: Optional[str] = None,
|
||||
*args,
|
||||
**kwargs,
|
||||
):
|
||||
"""Run the swarm on a given task with error handling and retries."""
|
||||
if not task or not task.strip():
|
||||
raise ValueError("Task cannot be empty")
|
||||
|
||||
logger.info("Starting swarm execution", extra={"task": task})
|
||||
|
||||
try:
|
||||
# Create agents for the task
|
||||
agents = self._create_agents(task)
|
||||
if not agents:
|
||||
raise ValueError(
|
||||
"No agents were created for the task"
|
||||
)
|
||||
|
||||
# Execute the task through the swarm router
|
||||
logger.info(
|
||||
"Routing task through swarm",
|
||||
extra={"num_agents": len(agents)},
|
||||
)
|
||||
output = self.swarm_router(
|
||||
agents=agents,
|
||||
task=task,
|
||||
image_url=image_url,
|
||||
*args,
|
||||
**kwargs,
|
||||
)
|
||||
self.auto_swarm_builder_output.swarm_router_outputs.append(
|
||||
output
|
||||
)
|
||||
print(output)
|
||||
|
||||
logger.info("Swarm execution completed successfully")
|
||||
# return output
|
||||
return self.auto_swarm_builder_output.model_dump_json(
|
||||
indent=4
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error during swarm execution: {str(e)}",
|
||||
)
|
||||
raise e
|
||||
|
||||
def _create_agents(
|
||||
self,
|
||||
task: str,
|
||||
) -> List[Agent]:
|
||||
"""Create the necessary agents for a task with enhanced error handling."""
|
||||
logger.info("Creating agents for task", extra={"task": task})
|
||||
|
||||
try:
|
||||
model = OpenAIFunctionCaller(
|
||||
system_prompt=BOSS_SYSTEM_PROMPT,
|
||||
api_key=self.api_key,
|
||||
temperature=0.1,
|
||||
base_model=SwarmConfig,
|
||||
)
|
||||
|
||||
agents_config = model.run(task)
|
||||
logger.info(
|
||||
f"Director has successfully created agents: {agents_config}"
|
||||
)
|
||||
self.auto_swarm_builder_output.director_agents_created = (
|
||||
agents_config
|
||||
)
|
||||
|
||||
if isinstance(agents_config, dict):
|
||||
agents_config = SwarmConfig(**agents_config)
|
||||
|
||||
# Update swarm configuration
|
||||
self.name = agents_config.name
|
||||
self.description = agents_config.description
|
||||
|
||||
# Create agents from configuration
|
||||
agents = []
|
||||
for agent_config in agents_config.agents:
|
||||
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,
|
||||
)
|
||||
agents.append(agent)
|
||||
|
||||
print(
|
||||
f"Agent created: {agent_config.name}: Description: {agent_config.description}"
|
||||
)
|
||||
|
||||
# # Add available agents showcase to system prompts
|
||||
# agents_available = showcase_available_agents(
|
||||
# name=self.name,
|
||||
# description=self.description,
|
||||
# agents=agents,
|
||||
# )
|
||||
|
||||
# for agent in agents:
|
||||
# agent.system_prompt += "\n" + agents_available
|
||||
|
||||
logger.info(
|
||||
"Successfully created agents",
|
||||
extra={"num_agents": len(agents)},
|
||||
)
|
||||
return agents
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error creating agents: {str(e)}", exc_info=True
|
||||
)
|
||||
raise
|
||||
|
||||
def build_agent(
|
||||
self,
|
||||
agent_name: str,
|
||||
agent_description: str,
|
||||
agent_system_prompt: str,
|
||||
*args,
|
||||
**kwargs,
|
||||
) -> Agent:
|
||||
"""Build a single agent with enhanced error handling."""
|
||||
logger.info(
|
||||
"Building agent", extra={"agent_name": agent_name}
|
||||
)
|
||||
|
||||
try:
|
||||
agent = Agent(
|
||||
agent_name=agent_name,
|
||||
description=agent_description,
|
||||
system_prompt=agent_system_prompt,
|
||||
model_name="gpt-4o",
|
||||
verbose=self.verbose,
|
||||
dynamic_temperature_enabled=False,
|
||||
return_step_meta=False,
|
||||
output_type="str",
|
||||
streaming_on=True,
|
||||
)
|
||||
return agent
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error building agent: {str(e)}", exc_info=True
|
||||
)
|
||||
raise
|
||||
|
||||
@retry(
|
||||
stop=stop_after_attempt(3),
|
||||
wait=wait_exponential(multiplier=1, min=4, max=10),
|
||||
)
|
||||
def swarm_router(
|
||||
self,
|
||||
agents: List[Agent],
|
||||
task: str,
|
||||
img: Optional[str] = None,
|
||||
*args,
|
||||
**kwargs,
|
||||
) -> str:
|
||||
"""Route tasks between agents in the swarm with error handling and retries."""
|
||||
logger.info(
|
||||
"Initializing swarm router",
|
||||
extra={"num_agents": len(agents)},
|
||||
)
|
||||
|
||||
try:
|
||||
swarm_router_instance = SwarmRouter(
|
||||
name=self.name,
|
||||
description=self.description,
|
||||
agents=agents,
|
||||
swarm_type=self.swarm_type,
|
||||
auto_generate_prompts=self.auto_generate_prompts_for_agents,
|
||||
)
|
||||
|
||||
# formatted_task = f"{self.name} {self.description} {task}"
|
||||
result = swarm_router_instance.run(
|
||||
task=task, *args, **kwargs
|
||||
)
|
||||
|
||||
logger.info("Successfully completed swarm routing")
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error in swarm router: {str(e)}", exc_info=True
|
||||
)
|
||||
raise
|
||||
|
||||
|
||||
# swarm = AutoSwarmBuilder(
|
||||
# name="ChipDesign-Swarm",
|
||||
# description="A swarm of specialized AI agents for chip design",
|
||||
# swarm_type="ConcurrentWorkflow",
|
||||
# )
|
||||
|
||||
# try:
|
||||
# result = swarm.run(
|
||||
# "Design a new AI accelerator chip optimized for transformer model inference..."
|
||||
# )
|
||||
# print(result)
|
||||
# except Exception as e:
|
||||
# print(f"An error occurred: {e}")
|
||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,64 @@
|
||||
import secrets
|
||||
import string
|
||||
import re
|
||||
|
||||
|
||||
def generate_api_key(prefix: str = "sk-", length: int = 32) -> str:
|
||||
"""
|
||||
Generate a secure API key with a custom prefix.
|
||||
|
||||
Args:
|
||||
prefix (str): The prefix to add to the API key (default: "sk-")
|
||||
length (int): The length of the random part of the key (default: 32)
|
||||
|
||||
Returns:
|
||||
str: A secure API key in the format: prefix + random_string
|
||||
|
||||
Example:
|
||||
>>> generate_api_key("sk-")
|
||||
'sk-abc123...'
|
||||
"""
|
||||
# Validate prefix
|
||||
if not isinstance(prefix, str):
|
||||
raise ValueError("Prefix must be a string")
|
||||
|
||||
# Validate length
|
||||
if length < 8:
|
||||
raise ValueError("Length must be at least 8 characters")
|
||||
|
||||
# Generate random string using alphanumeric characters
|
||||
alphabet = string.ascii_letters + string.digits
|
||||
random_part = "".join(
|
||||
secrets.choice(alphabet) for _ in range(length)
|
||||
)
|
||||
|
||||
# Combine prefix and random part
|
||||
api_key = f"{prefix}{random_part}"
|
||||
|
||||
return api_key
|
||||
|
||||
|
||||
def validate_api_key(api_key: str, prefix: str = "sk-") -> bool:
|
||||
"""
|
||||
Validate if an API key matches the expected format.
|
||||
|
||||
Args:
|
||||
api_key (str): The API key to validate
|
||||
prefix (str): The expected prefix (default: "sk-")
|
||||
|
||||
Returns:
|
||||
bool: True if the API key is valid, False otherwise
|
||||
"""
|
||||
if not isinstance(api_key, str):
|
||||
return False
|
||||
|
||||
# Check if key starts with prefix
|
||||
if not api_key.startswith(prefix):
|
||||
return False
|
||||
|
||||
# Check if the rest of the key contains only alphanumeric characters
|
||||
random_part = api_key[len(prefix) :]
|
||||
if not re.match(r"^[a-zA-Z0-9]+$", random_part):
|
||||
return False
|
||||
|
||||
return True
|
||||
Loading…
Reference in new issue