ai-mirow-projects
/
swarms
mirror of https://github.com/kyegomez/swarms
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactor error handling and improve logging

Browse Source
pull/1231/head
CI-DEV 4 weeks ago committed by GitHub
parent fb11d5d972
commit 8f6762ba15
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 296 additions and 109 deletions
Show Stats Download Patch File Download Diff File

405
swarms/structs/hiearchical_swarm.py
Unescape View File

@ -16,10 +16,16 @@ Todo
- Add layers of management -- a list of list of agents that act as departments
- Auto build agents from input prompt - and then add them to the swarm
- Create an interactive and dynamic UI like we did with heavy swarm
- Make it faster and more high performance
- Enable the director to choose a multi-agent approach to the task, it orchestrates how the agents talk and work together.
- Improve the director feedback, maybe add agent as a judge to the worker agent instead of the director.
- Use agent rearrange to orchestrate the agents
Classes:
HierarchicalOrder: Represents a single task assignment to a specific agent
SwarmSpec: Contains the overall plan and list of orders for the swarm
HierarchicalSwarm: Main swarm orchestrator that manages director and worker agents
"""
import time
@ -37,11 +43,11 @@ from rich.text import Text
from swarms.prompts.hiearchical_system_prompt import (
HIEARCHICAL_SWARM_SYSTEM_PROMPT,
DIRECTOR_PLANNING_PROMPT,
)
from swarms.prompts.multi_agent_collab_prompt import (
MULTI_AGENT_COLLAB_PROMPT_TWO,
)
from swarms.prompts.reasoning_prompt import INTERNAL_MONOLGUE_PROMPT
from swarms.structs.agent import Agent
from swarms.structs.conversation import Conversation
from swarms.structs.ma_utils import list_all_agents
@ -51,6 +57,7 @@ from swarms.utils.history_output_formatter import (
history_output_formatter,
)
from swarms.utils.output_types import OutputType
from swarms.utils.formatter import formatter
class HierarchicalSwarmDashboard:
@ -612,6 +619,13 @@ class SwarmSpec(BaseModel):
individual agents within the swarm.
"""
# # thoughts: str = Field(
# # ...,
# # description="A plan generated by the director agent for the swarm to accomplish the given task, where the director autonomously reasons through the problem, devises its own strategy, and determines the sequence of actions. "
# # "This plan reflects the director's independent thought process, outlining the rationale, priorities, and steps it deems necessary for successful execution. "
# # "It serves as a blueprint for the swarm, enabling agents to follow the director's self-derived guidance and adapt as needed throughout the process.",
# )
plan: str = Field(
...,
description="A plan generated by the director agent for the swarm to accomplish the given task, where the director autonomously reasons through the problem, devises its own strategy, and determines the sequence of actions. "
@ -648,7 +662,10 @@ class HierarchicalSwarm:
feedback_director_model_name (str): Model name for the feedback director.
director_name (str): Name identifier for the director agent.
director_model_name (str): Model name for the main director agent.
verbose (bool): Whether to enable detailed logging and progress tracking.
add_collaboration_prompt (bool): Whether to add collaboration prompts to agents.
planning_director_agent (Optional[Union[Agent, Callable, Any]]): Optional
planning agent.
director_feedback_on (bool): Whether director feedback is enabled.
"""
@ -663,14 +680,18 @@ class HierarchicalSwarm:
feedback_director_model_name: str = "gpt-4o-mini",
director_name: str = "Director",
director_model_name: str = "gpt-4o-mini",
verbose: bool = False,
add_collaboration_prompt: bool = True,
planning_director_agent: Optional[
Union[Agent, Callable, Any]
] = None,
director_feedback_on: bool = True,
interactive: bool = False,
director_system_prompt: str = HIEARCHICAL_SWARM_SYSTEM_PROMPT,
director_reasoning_model_name: str = "o3-mini",
director_reasoning_enabled: bool = False,
multi_agent_prompt_improvements: bool = False,
director_temperature: float = 0.7,
director_top_p: float = 0.9,
planning_enabled: bool = True,
show_swarm_structure: bool = False,
*args,
**kwargs,
):
@ -689,8 +710,12 @@ class HierarchicalSwarm:
feedback_director_model_name (str): Model name for feedback director.
director_name (str): Name identifier for the director agent.
director_model_name (str): Model name for the main director agent.
verbose (bool): Whether to enable detailed logging.
add_collaboration_prompt (bool): Whether to add collaboration prompts.
planning_director_agent (Optional[Union[Agent, Callable, Any]]):
Optional planning agent for enhanced planning capabilities.
director_feedback_on (bool): Whether director feedback is enabled.
show_swarm_structure (bool): Whether to automatically print the nested swarm structure. Defaults to False.
*args: Additional positional arguments.
**kwargs: Additional keyword arguments.
@ -707,17 +732,21 @@ class HierarchicalSwarm:
feedback_director_model_name
)
self.director_name = director_name
self.verbose = verbose
self.director_model_name = director_model_name
self.add_collaboration_prompt = add_collaboration_prompt
self.planning_director_agent = planning_director_agent
self.director_feedback_on = director_feedback_on
self.interactive = interactive
self.director_system_prompt = director_system_prompt
self.director_reasoning_model_name = (
director_reasoning_model_name
)
self.director_reasoning_enabled = director_reasoning_enabled
self.multi_agent_prompt_improvements = (
multi_agent_prompt_improvements
)
self.director_temperature = director_temperature
self.director_top_p = director_top_p
self.planning_enabled = planning_enabled
self.show_swarm_structure = show_swarm_structure
self.initialize_swarm()
@ -759,6 +788,33 @@ class HierarchicalSwarm:
else:
agent.system_prompt = prompt
def reasoning_agent_run(
self, task: str, img: Optional[str] = None
):
"""
Run a reasoning agent to analyze the task before the main director processes it.
Args:
task (str): The task to reason about
img (Optional[str]): Optional image input
Returns:
str: The reasoning output from the agent
"""
agent = Agent(
agent_name=self.director_name,
agent_description=f"You're the {self.director_name} agent that is responsible for reasoning about the task and creating a plan for the swarm to accomplish the task.",
model_name=self.director_reasoning_model_name,
system_prompt=INTERNAL_MONOLGUE_PROMPT
+ self.director_system_prompt,
max_loops=1,
)
prompt = f"Conversation History: {self.conversation.get_str()} \n\n Task: {task}"
return agent.run(task=prompt, img=img)
def init_swarm(self):
"""
Initialize the swarm with proper configuration and validation.
@ -772,12 +828,17 @@ class HierarchicalSwarm:
Raises:
ValueError: If the swarm configuration is invalid.
"""
# Initialize logger only if verbose is enabled
if self.verbose:
logger.info(
f"[INIT] Initializing HierarchicalSwarm: {self.name}"
)
self.conversation = Conversation(time_enabled=False)
# Reliability checks
self.reliability_checks()
# Add agent context to the director
self.add_context_to_director()
# Initialize agent statuses in dashboard if interactive mode
@ -793,8 +854,29 @@ class HierarchicalSwarm:
# Force refresh to ensure agents are displayed
self.dashboard.force_refresh()
if self.verbose:
logger.success(
f"[SUCCESS] HierarchicalSwarm: {self.name} initialized successfully."
)
if self.multi_agent_prompt_improvements:
self.prepare_worker_agents()
# Print swarm structure if enabled
if self.show_swarm_structure:
try:
formatter.print_swarm_structure(self)
except Exception as e:
if self.verbose:
logger.warning(f"Failed to print swarm structure: {e}")
def print_swarm_structure(self) -> None:
"""
Print the visual tree representation of this swarm's structure.
This is a convenience method that calls the formatter's print_swarm_structure method.
"""
formatter.print_swarm_structure(self)
def add_context_to_director(self):
"""
@ -809,6 +891,9 @@ class HierarchicalSwarm:
Exception: If adding context fails due to agent configuration issues.
"""
try:
if self.verbose:
logger.info("[INFO] Adding agent context to director")
list_all_agents(
agents=self.agents,
conversation=self.conversation,
@ -816,6 +901,11 @@ class HierarchicalSwarm:
add_collaboration_prompt=self.add_collaboration_prompt,
)
if self.verbose:
logger.success(
"[SUCCESS] Agent context added to director successfully"
)
except Exception as e:
error_msg = (
f"[ERROR] Failed to add context to director: {str(e)}"
@ -838,15 +928,19 @@ class HierarchicalSwarm:
Exception: If director setup fails due to configuration issues.
"""
try:
if self.verbose:
logger.info("[SETUP] Setting up director agent")
schema = BaseTool().base_model_to_dict(SwarmSpec)
if self.verbose:
logger.debug(f"[SCHEMA] Director schema: {schema}")
return Agent(
agent_name=self.director_name,
agent_description="A director agent that can create a plan and distribute orders to agents",
system_prompt=self.director_system_prompt,
model_name=self.director_model_name,
temperature=self.director_temperature,
top_p=self.director_top_p,
max_loops=1,
base_model=SwarmSpec,
tools_list_dictionary=[schema],
@ -854,34 +948,8 @@ class HierarchicalSwarm:
)
except Exception as e:
error_msg = f"[ERROR] Failed to setup director: {str(e)}"
logger.error(
f"{error_msg}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
)
def setup_director_with_planning(
self, task: str = None, img: Optional[str] = None
):
try:
agent = Agent(
agent_name=self.director_name,
agent_description="A director agent that can create a plan and distribute orders to agents",
system_prompt=DIRECTOR_PLANNING_PROMPT,
model_name=self.director_model_name,
temperature=self.director_temperature,
top_p=self.director_top_p,
max_loops=1,
output_type="final",
)
return agent.run(task=task, img=img)
except Exception as e:
error_msg = f"[ERROR] Failed to setup director with planning: {str(e)}"
logger.error(
f"{error_msg}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
)
error_msg = f"[ERROR] Failed to setup director: {str(e)}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
logger.error(error_msg)
def reliability_checks(self):
"""
@ -896,6 +964,11 @@ class HierarchicalSwarm:
ValueError: If the swarm configuration is invalid.
"""
try:
if self.verbose:
logger.info(
f"Hiearchical Swarm: {self.name} Reliability checks in progress..."
)
if not self.agents or len(self.agents) == 0:
raise ValueError(
"No agents found in the swarm. At least one agent must be provided to create a hierarchical swarm."
@ -909,12 +982,14 @@ class HierarchicalSwarm:
if self.director is None:
self.director = self.setup_director()
if self.verbose:
logger.success(
f"Hiearchical Swarm: {self.name} Reliability checks passed..."
)
except Exception as e:
error_msg = f"[ERROR] Reliability checks failed: {str(e)}"
logger.error(
f"{error_msg}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
)
raise e
error_msg = f"[ERROR] Failed to setup director: {str(e)}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
logger.error(error_msg)
def agents_no_print(self):
for agent in self.agents:
@ -929,7 +1004,9 @@ class HierarchicalSwarm:
Execute the director agent with the given task and conversation context.
This method runs the director agent to create a plan and distribute orders
based on the current task and conversation history.
based on the current task and conversation history. If a planning director
agent is configured, it will first create a detailed plan before the main
director processes the task.
Args:
task (str): The task to be executed by the director.
@ -942,15 +1019,24 @@ class HierarchicalSwarm:
Exception: If director execution fails.
"""
try:
if self.planning_enabled is True:
self.director.tools_list_dictionary = None
out = self.setup_director_with_planning(
task=f"History: {self.conversation.get_str()} \n\n Task: {task}",
if self.verbose:
logger.info(
f"[RUN] Running director with task: {task}"
)
if self.planning_director_agent is not None:
plan = self.planning_director_agent.run(
task=f"History: {self.conversation.get_str()} \n\n Create a detailed step by step comprehensive plan for the director to execute the task: {task}",
img=img,
)
self.conversation.add(
role=self.director.agent_name, content=out
task += plan
if self.director_reasoning_enabled:
reasoning_output = self.reasoning_agent_run(
task=task, img=img
)
task += f"\n\n Reasoning: {reasoning_output}"
# Run the director with the context
function_call = self.director.run(
@ -962,13 +1048,19 @@ class HierarchicalSwarm:
role="Director", content=function_call
)
if self.verbose:
logger.success(
"[SUCCESS] Director execution completed"
)
logger.debug(
f"[OUTPUT] Director output type: {type(function_call)}"
)
return function_call
except Exception as e:
error_msg = f"[ERROR] Failed to run director: {str(e)}"
logger.error(
f"{error_msg}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
)
error_msg = f"[ERROR] Failed to setup director: {str(e)}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
logger.error(error_msg)
raise e
def step(
@ -1006,6 +1098,11 @@ class HierarchicalSwarm:
Exception: If step execution fails.
"""
try:
if self.verbose:
logger.info(
f"[STEP] Executing single step for task: {task}"
)
# Update dashboard for director execution
if self.interactive and self.dashboard:
self.dashboard.update_director_status("PLANNING")
@ -1015,6 +1112,11 @@ class HierarchicalSwarm:
# Parse the orders
plan, orders = self.parse_orders(output)
if self.verbose:
logger.info(
f"[PARSE] Parsed plan and {len(orders)} orders"
)
# Update dashboard with plan and orders information
if self.interactive and self.dashboard:
self.dashboard.update_director_plan(plan)
@ -1034,18 +1136,24 @@ class HierarchicalSwarm:
orders, streaming_callback=streaming_callback
)
if self.verbose:
logger.info(f"[EXEC] Executed {len(outputs)} orders")
if self.director_feedback_on is True:
feedback = self.feedback_director(outputs)
else:
feedback = outputs
if self.verbose:
logger.success(
"[SUCCESS] Step completed successfully"
)
return feedback
except Exception as e:
error_msg = f"[ERROR] Step execution failed: {str(e)}"
logger.error(
f"{error_msg}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
)
error_msg = f"[ERROR] Failed to setup director: {str(e)}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
logger.error(error_msg)
def run(
self,
@ -1097,7 +1205,20 @@ class HierarchicalSwarm:
self.dashboard.start(self.max_loops)
self.dashboard.update_director_status("ACTIVE")
if self.verbose:
logger.info(
f"[START] Starting hierarchical swarm run: {self.name}"
)
logger.info(
f"[CONFIG] Configuration - Max loops: {self.max_loops}"
)
while current_loop < self.max_loops:
if self.verbose:
logger.info(
f"[LOOP] Loop {current_loop + 1}/{self.max_loops} - Processing task"
)
# Update dashboard loop counter
if self.interactive and self.dashboard:
self.dashboard.update_loop(current_loop + 1)
@ -1127,13 +1248,14 @@ class HierarchicalSwarm:
**kwargs,
)
if self.verbose:
logger.success(
f"[SUCCESS] Loop {current_loop + 1} completed successfully"
)
except Exception as e:
error_msg = (
f"[ERROR] Loop execution failed: {str(e)}"
)
logger.error(
f"{error_msg}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
)
error_msg = f"[ERROR] Failed to setup director: {str(e)}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
logger.error(error_msg)
current_loop += 1
@ -1148,6 +1270,14 @@ class HierarchicalSwarm:
self.dashboard.update_director_status("COMPLETED")
self.dashboard.stop()
if self.verbose:
logger.success(
f"[COMPLETE] Hierarchical swarm run completed: {self.name}"
)
logger.info(
f"[STATS] Total loops executed: {current_loop}"
)
return history_output_formatter(
conversation=self.conversation, type=self.output_type
)
@ -1158,10 +1288,8 @@ class HierarchicalSwarm:
self.dashboard.update_director_status("ERROR")
self.dashboard.stop()
error_msg = f"[ERROR] Swarm run failed: {str(e)}"
logger.error(
f"{error_msg}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
)
error_msg = f"[ERROR] Failed to setup director: {str(e)}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
logger.error(error_msg)
def _get_interactive_task(self) -> str:
"""
@ -1200,6 +1328,9 @@ class HierarchicalSwarm:
Exception: If feedback generation fails.
"""
try:
if self.verbose:
logger.info("[FEEDBACK] Generating director feedback")
task = f"History: {self.conversation.get_str()} \n\n"
feedback_director = Agent(
@ -1223,13 +1354,16 @@ class HierarchicalSwarm:
role=self.director.agent_name, content=output
)
if self.verbose:
logger.success(
"[SUCCESS] Director feedback generated successfully"
)
return output
except Exception as e:
error_msg = f"[ERROR] Feedback director failed: {str(e)}"
logger.error(
f"{error_msg}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
)
error_msg = f"[ERROR] Failed to setup director: {str(e)}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
logger.error(error_msg)
def call_single_agent(
self,
@ -1265,6 +1399,9 @@ class HierarchicalSwarm:
Exception: If agent execution fails.
"""
try:
if self.verbose:
logger.info(f"[CALL] Calling agent: {agent_name}")
# Find agent by name
agent = None
for a in self.agents:
@ -1301,11 +1438,11 @@ class HierarchicalSwarm:
streaming_callback(
agent_name, chunk, False
)
except Exception as e:
error_msg = f"[ERROR] Streaming callback failed for agent {agent_name}: {str(e)}"
logger.error(
f"{error_msg}\n[TRACE] Traceback: {traceback.format_exc()}"
)
except Exception as callback_error:
if self.verbose:
logger.warning(
f"[STREAMING] Callback failed for {agent_name}: {str(callback_error)}"
)
output = agent.run(
task=f"History: {self.conversation.get_str()} \n\n Task: {task}",
@ -1317,11 +1454,11 @@ class HierarchicalSwarm:
# Call completion callback
try:
streaming_callback(agent_name, "", True)
except Exception as e:
error_msg = f"[ERROR] Completion callback failed for agent {agent_name}: {str(e)}"
logger.error(
f"{error_msg}\n[TRACE] Traceback: {traceback.format_exc()}"
)
except Exception as callback_error:
if self.verbose:
logger.warning(
f"[STREAMING] Completion callback failed for {agent_name}: {str(callback_error)}"
)
else:
output = agent.run(
task=f"History: {self.conversation.get_str()} \n\n Task: {task}",
@ -1330,6 +1467,11 @@ class HierarchicalSwarm:
)
self.conversation.add(role=agent_name, content=output)
if self.verbose:
logger.success(
f"[SUCCESS] Agent {agent_name} completed task successfully"
)
return output
except Exception as e:
@ -1339,12 +1481,8 @@ class HierarchicalSwarm:
agent_name, "ERROR", task, f"Error: {str(e)}"
)
error_msg = (
f"[ERROR] Failed to call agent {agent_name}: {str(e)}"
)
logger.error(
f"{error_msg}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
)
error_msg = f"[ERROR] Failed to setup director: {str(e)}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
logger.error(error_msg)
def parse_orders(self, output):
"""
@ -1366,6 +1504,10 @@ class HierarchicalSwarm:
Exception: If parsing fails due to other errors.
"""
try:
if self.verbose:
logger.info("[PARSE] Parsing director orders")
logger.debug(f"[TYPE] Output type: {type(output)}")
import json
# Handle different output formats from the director
@ -1406,8 +1548,19 @@ class HierarchicalSwarm:
]
]
if self.verbose:
logger.success(
f"[SUCCESS] Successfully parsed plan and {len(orders)} orders"
)
return plan, orders
except json.JSONDecodeError:
except (
json.JSONDecodeError
) as json_err:
if self.verbose:
logger.warning(
f"[WARN] JSON decode error: {json_err}"
)
pass
# Check if it's a direct function call format
elif "function" in item:
@ -1429,8 +1582,19 @@ class HierarchicalSwarm:
]
]
if self.verbose:
logger.success(
f"[SUCCESS] Successfully parsed plan and {len(orders)} orders"
)
return plan, orders
except json.JSONDecodeError:
except (
json.JSONDecodeError
) as json_err:
if self.verbose:
logger.warning(
f"[WARN] JSON decode error: {json_err}"
)
pass
# If no function call found, raise error
raise ValueError(
@ -1445,6 +1609,11 @@ class HierarchicalSwarm:
for order in output["orders"]
]
if self.verbose:
logger.success(
f"[SUCCESS] Successfully parsed plan and {len(orders)} orders"
)
return plan, orders
else:
raise ValueError(
@ -1456,10 +1625,8 @@ class HierarchicalSwarm:
)
except Exception as e:
error_msg = f"[ERROR] Failed to parse orders: {str(e)}"
logger.error(
f"{error_msg}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
)
error_msg = f"[ERROR] Failed to parse orders: {str(e)}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
logger.error(error_msg)
raise e
def execute_orders(
@ -1489,8 +1656,16 @@ class HierarchicalSwarm:
Exception: If order execution fails.
"""
try:
if self.verbose:
logger.info(f"[EXEC] Executing {len(orders)} orders")
outputs = []
for i, order in enumerate(orders):
if self.verbose:
logger.info(
f"[ORDER] Executing order {i+1}/{len(orders)}: {order.agent_name}"
)
# Update dashboard for agent execution
if self.interactive and self.dashboard:
self.dashboard.update_agent_status(
@ -1520,21 +1695,15 @@ class HierarchicalSwarm:
outputs.append(output)
if self.verbose:
logger.success(
f"[SUCCESS] All {len(orders)} orders executed successfully"
)
return outputs
except Exception as e:
error_msg = (
"\n"
+ "=" * 60
+ "\n[SWARMS ERROR] Order Execution Failure\n"
+ "-" * 60
+ f"\nError : {str(e)}"
f"\nTrace :\n{traceback.format_exc()}"
+ "-" * 60
+ "\nIf this issue persists, please report it:"
"\n https://github.com/kyegomez/swarms/issues"
"\n" + "=" * 60 + "\n"
)
error_msg = f"[ERROR] Failed to setup director: {str(e)}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
logger.error(error_msg)
def batched_run(
@ -1570,6 +1739,14 @@ class HierarchicalSwarm:
Exception: If batched execution fails.
"""
try:
if self.verbose:
logger.info(
f"[START] Starting batched hierarchical swarm run: {self.name}"
)
logger.info(
f"[CONFIG] Configuration - Max loops: {self.max_loops}"
)
# Initialize a list to store the results
results = []
@ -1584,10 +1761,20 @@ class HierarchicalSwarm:
)
results.append(result)
if self.verbose:
logger.success(
f"[COMPLETE] Batched hierarchical swarm run completed: {self.name}"
)
logger.info(
f"[STATS] Total tasks processed: {len(tasks)}"
)
return results
except Exception as e:
error_msg = f"[ERROR] Batched hierarchical swarm run failed: {str(e)}"
logger.error(
f"{error_msg}\n[TRACE] Traceback: {traceback.format_exc()}\n[BUG] If this issue persists, please report it at: https://github.com/kyegomez/swarms/issues"
)
if self.verbose:
logger.error(error_msg)
logger.error(
f"[TRACE] Traceback: {traceback.format_exc()}"
)

Write Preview
Loading…
Cancel
Save