diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml index 868cb092..c57ef306 100644 --- a/docs/mkdocs.yml +++ b/docs/mkdocs.yml @@ -187,6 +187,7 @@ nav: - Various Execution Methods: "swarms/structs/various_execution_methods.md" - Workflows: - ConcurrentWorkflow: "swarms/structs/concurrentworkflow.md" + - AsyncWorkflow: "swarms/structs/async_workflow.md" - SequentialWorkflow: "swarms/structs/sequential_workflow.md" - Structs: - Conversation: "swarms/structs/conversation.md" diff --git a/docs/swarms/structs/async_workflow.md b/docs/swarms/structs/async_workflow.md index 4f9657e7..f90ac059 100644 --- a/docs/swarms/structs/async_workflow.md +++ b/docs/swarms/structs/async_workflow.md @@ -1,266 +1,181 @@ # AsyncWorkflow Documentation -The `AsyncWorkflow` class represents an asynchronous workflow designed to execute tasks concurrently. This class is ideal for scenarios where tasks need to be run asynchronously, leveraging Python's asyncio capabilities to manage multiple tasks efficiently. +The `AsyncWorkflow` class represents an asynchronous workflow that executes tasks concurrently using multiple agents. It allows for efficient task management, leveraging Python's `asyncio` for concurrent execution. -### Key Concepts +## Key Features +- **Concurrent Task Execution**: Distribute tasks across multiple agents asynchronously. +- **Configurable Workers**: Limit the number of concurrent workers (agents) for better resource management. +- **Autosave Results**: Optionally save the task execution results automatically. +- **Verbose Logging**: Enable detailed logging to monitor task execution. +- **Error Handling**: Gracefully handles exceptions raised by agents during task execution. -- **Asynchronous Execution**: Tasks are run concurrently using asyncio, allowing for non-blocking operations. -- **Task Pool**: A collection of tasks to be executed within the workflow. -- **Event Loop**: The asyncio event loop that manages the execution of asynchronous tasks. -- **Stopping Condition**: A condition that, when met, stops the execution of the workflow. +--- ## Attributes - -### Arguments - -| Argument | Type | Default | Description | -|----------|------|---------|-------------| -| `name` | `str` | `"Async Workflow"` | The name of the workflow. | -| `description` | `str` | `"A workflow to run asynchronous tasks"` | The description of the workflow. | -| `max_loops` | `int` | `1` | The maximum number of loops to run the workflow. | -| `autosave` | `bool` | `True` | Flag indicating whether to autosave the results. | -| `dashboard` | `bool` | `False` | Flag indicating whether to display a dashboard. | -| `task_pool` | `List[Any]` | `[]` | The list of tasks in the workflow. | -| `results` | `List[Any]` | `[]` | The list of results from running the tasks. | -| `loop` | `Optional[asyncio.AbstractEventLoop]` | `None` | The event loop to use. | -| `stopping_condition` | `Optional[Callable]` | `None` | The stopping condition for the workflow. | -| `agents` | `List[Agent]` | `None` | A list of agents participating in the workflow. | - -### Attributes - -| Attribute | Type | Description | -|-----------|------|-------------| -| `name` | `str` | The name of the workflow. | -| `description` | `str` | The description of the workflow. | -| `max_loops` | `int` | The maximum number of loops to run the workflow. | -| `autosave` | `bool` | Flag indicating whether to autosave the results. | -| `dashboard` | `bool` | Flag indicating whether to display a dashboard. | -| `task_pool` | `List[Any]` | The list of tasks in the workflow. | -| `results` | `List[Any]` | The list of results from running the tasks. | -| `loop` | `Optional[asyncio.AbstractEventLoop]` | The event loop to use. | -| `stopping_condition` | `Optional[Callable]` | The stopping condition for the workflow. | -| `agents` | `List[Agent]` | A list of agents participating in the workflow. | - -## Methods - -### add - -Adds a task or a list of tasks to the task pool. - -**Arguments:** - -| Parameter | Type | Default | Description | -|-----------|------|---------|-------------| -| `task` | `Any` | `None` | A single task to add. | -| `tasks` | `List[Any]` | `None` | A list of tasks to add. | - -**Raises:** - -- `ValueError`: If neither task nor tasks are provided. - -**Examples:** - +| Attribute | Type | Description | +|-------------------|---------------------|-----------------------------------------------------------------------------| +| `name` | `str` | The name of the workflow. | +| `agents` | `List[Agent]` | A list of agents participating in the workflow. | +| `max_workers` | `int` | The maximum number of concurrent workers (default: 5). | +| `dashboard` | `bool` | Whether to display a dashboard (currently not implemented). | +| `autosave` | `bool` | Whether to autosave task results (default: `False`). | +| `verbose` | `bool` | Whether to enable detailed logging (default: `False`). | +| `task_pool` | `List` | A pool of tasks to be executed. | +| `results` | `List` | A list to store results of executed tasks. | +| `loop` | `asyncio.EventLoop` | The event loop for asynchronous execution. | + +--- + +**Description**: +Initializes the `AsyncWorkflow` with specified agents, configuration, and options. + +**Parameters**: +- `name` (`str`): Name of the workflow. Default: "AsyncWorkflow". +- `agents` (`List[Agent]`): A list of agents. Default: `None`. +- `max_workers` (`int`): The maximum number of workers. Default: `5`. +- `dashboard` (`bool`): Enable dashboard visualization (placeholder for future implementation). +- `autosave` (`bool`): Enable autosave of task results. Default: `False`. +- `verbose` (`bool`): Enable detailed logging. Default: `False`. +- `**kwargs`: Additional parameters for `BaseWorkflow`. + +--- + +### `_execute_agent_task` ```python -workflow = AsyncWorkflow() -task1 = Task(description="Task 1") -task2 = Task(description="Task 2") - -# Adding a single task -await workflow.add(task=task1) - -# Adding multiple tasks -await workflow.add(tasks=[task1, task2]) +async def _execute_agent_task(self, agent: Agent, task: str) -> Any: ``` +**Description**: +Executes a single task asynchronously using a given agent. -### delete +**Parameters**: +- `agent` (`Agent`): The agent responsible for executing the task. +- `task` (`str`): The task to be executed. -Deletes a task from the workflow. - -**Arguments:** - -| Parameter | Type | Default | Description | -|-----------|------|---------|-------------| -| `task` | `Any` | `None` | A single task to delete. | -| `tasks` | `List[Task]` | `None` | A list of tasks to delete. | - -**Examples:** +**Returns**: +- `Any`: The result of the task execution or an error message in case of an exception. +**Example**: ```python -workflow = AsyncWorkflow() -task1 = Task(description="Task 1") -task2 = Task(description="Task 2") - -# Adding tasks to the workflow -await workflow.add(tasks=[task1, task2]) - -# Deleting a single task -await workflow.delete(task=task1) - -# Deleting multiple tasks -await workflow.delete(tasks=[task1, task2]) +result = await workflow._execute_agent_task(agent, "Sample Task") ``` -### run - -Runs the workflow and returns the results. - -**Returns:** - -| Return Type | Description | -|-------------|-------------| -| `List[Any]` | The results of the executed tasks. | - -**Examples:** +--- +### `run` ```python -workflow = AsyncWorkflow() -task1 = Task(description="Task 1", execute=async_function) -task2 = Task(description="Task 2", execute=async_function) - -# Adding tasks to the workflow -await workflow.add(tasks=[task1, task2]) - -# Running the workflow -results = await workflow.run() +async def run(self, task: str) -> List[Any]: ``` +**Description**: +Executes the specified task concurrently across all agents. -### Additional Examples - -#### Example 1: Simple AsyncWorkflow - -```python -import asyncio -from swarms.structs.agent import Agent -from swarms.structs.task import Task - -async def simple_task(): - await asyncio.sleep(1) - return "Task Completed" - -workflow = AsyncWorkflow() -task = Task(description="Simple Task", execute=simple_task) +**Parameters**: +- `task` (`str`): The task to be executed by all agents. -# Adding a task to the workflow -await workflow.add(task=task) +**Returns**: +- `List[Any]`: A list of results or error messages returned by the agents. -# Running the workflow -results = await workflow.run() -print(results) # Output: ["Task Completed"] -``` - -#### Example 2: Workflow with Multiple Tasks +**Raises**: +- `ValueError`: If no agents are provided in the workflow. +**Example**: ```python import asyncio -from swarms.structs.agent import Agent -from swarms.structs.task import Task - -async def task1(): - await asyncio.sleep(1) - return "Task 1 Completed" - -async def task2(): - await asyncio.sleep(2) - return "Task 2 Completed" -workflow = AsyncWorkflow() -task_1 = Task(description="Task 1", execute=task1) -task_2 = Task(description="Task 2", execute=task2) +agents = [Agent("Agent1"), Agent("Agent2")] +workflow = AsyncWorkflow(agents=agents, verbose=True) -# Adding tasks to the workflow -await workflow.add(tasks=[task_1, task_2]) - -# Running the workflow -results = await workflow.run() -print(results) # Output: ["Task 1 Completed", "Task 2 Completed"] +results = asyncio.run(workflow.run("Process Data")) +print(results) ``` -#### Example 3: Workflow with Stopping Condition +--- +## Production-Grade Financial Example: Multiple Agents +### Example: Stock Analysis and Investment Strategy ```python import asyncio -from swarms.structs.agent import Agent -from swarms.structs.task import Task - -async def task1(): - await asyncio.sleep(1) - return "Task 1 Completed" +from swarms import Agent +from async_workflow import AsyncWorkflow +from swarms.prompts.finance_agent_sys_prompt import FINANCIAL_AGENT_SYS_PROMPT + +# Initialize multiple Financial Agents +portfolio_analysis_agent = Agent( + agent_name="Portfolio-Analysis-Agent", + system_prompt=FINANCIAL_AGENT_SYS_PROMPT, + model_name="gpt-4o-mini", + autosave=True, + verbose=True, +) -async def task2(): - await asyncio.sleep(2) - return "Task 2 Completed" +stock_strategy_agent = Agent( + agent_name="Stock-Strategy-Agent", + system_prompt=FINANCIAL_AGENT_SYS_PROMPT, + model_name="gpt-4o-mini", + autosave=True, + verbose=True, +) -def stop_condition(results): - return "Task 2 Completed" in results +risk_management_agent = Agent( + agent_name="Risk-Management-Agent", + system_prompt=FINANCIAL_AGENT_SYS_PROMPT, + model_name="gpt-4o-mini", + autosave=True, + verbose=True, +) -workflow = AsyncWorkflow(stopping_condition=stop_condition) -task_1 = Task(description="Task 1", execute=task1) -task_2 = Task(description="Task 2", execute=task2) +# Create a workflow with multiple agents +workflow = AsyncWorkflow( + name="Financial-Workflow", + agents=[portfolio_analysis_agent, stock_strategy_agent, risk_management_agent], + verbose=True, +) -# Adding tasks to the workflow -await workflow.add(tasks=[task_1, task_2]) +# Run the workflow +async def main(): + task = "Analyze the current stock market trends and provide an investment strategy with risk assessment." + results = await workflow.run(task) + for agent_result in results: + print(agent_result) -# Running the workflow -results = await workflow.run() -print(results) # Output: ["Task 1 Completed", "Task 2 Completed"] +asyncio.run(main()) ``` -# Async Workflow +**Output**: +``` +INFO: Agent Portfolio-Analysis-Agent processing task: Analyze the current stock market trends and provide an investment strategy with risk assessment. +INFO: Agent Stock-Strategy-Agent processing task: Analyze the current stock market trends and provide an investment strategy with risk assessment. +INFO: Agent Risk-Management-Agent processing task: Analyze the current stock market trends and provide an investment strategy with risk assessment. +INFO: Agent Portfolio-Analysis-Agent completed task +INFO: Agent Stock-Strategy-Agent completed task +INFO: Agent Risk-Management-Agent completed task +Results: +- Detailed portfolio analysis... +- Stock investment strategies... +- Risk assessment insights... +``` -The AsyncWorkflow allows multiple agents to process tasks concurrently using Python's asyncio framework. +--- -## Usage Example +## Notes +1. **Autosave**: The autosave functionality is a placeholder. Users can implement custom logic to save `self.results`. +2. **Error Handling**: Exceptions raised by agents are logged and returned as part of the results. +3. **Dashboard**: The `dashboard` feature is currently not implemented but can be extended for visualization. -```python -import asyncio -from swarms import Agent, AsyncWorkflow -from swarm_models import OpenAIChat - -# Initialize model -model = OpenAIChat( - openai_api_key="your-api-key", - model_name="gpt-4", - temperature=0.7 -) +--- -# Create agents -agents = [ - Agent( - agent_name=f"Analysis-Agent-{i}", - llm=model, - max_loops=1, - dashboard=False, - verbose=True, - ) - for i in range(3) -] - -# Initialize workflow -workflow = AsyncWorkflow( - name="Analysis-Workflow", - agents=agents, - max_workers=3, - verbose=True -) +## Dependencies +- `asyncio`: Python's asynchronous I/O framework. +- `loguru`: Logging utility for better log management. +- `swarms`: Base components (`BaseWorkflow`, `Agent`). -# Run workflow -async def main(): - task = "Analyze the potential impact of AI on healthcare" - results = await workflow.run(task) - for i, result in enumerate(results): - print(f"Agent {i} result: {result}") +--- -# Execute -asyncio.run(main()) -``` +## Future Extensions +- **Dashboard**: Implement a real-time dashboard for monitoring agent performance. +- **Autosave**: Add persistent storage support for task results. +- **Task Management**: Extend task pooling and scheduling logic to support dynamic workloads. -## Parameters +--- -| Parameter | Type | Default | Description | -|-----------|------|---------|-------------| -| `name` | str | "AsyncWorkflow" | Name of the workflow | -| `agents` | List[Agent] | None | List of agents to execute tasks | -| `max_workers` | int | 5 | Maximum number of concurrent workers | -| `dashboard` | bool | False | Enable/disable dashboard | -| `autosave` | bool | False | Enable/disable autosaving results | -| `verbose` | bool | False | Enable/disable verbose logging | \ No newline at end of file +## License +This class is part of the `swarms` framework and follows the framework's licensing terms. diff --git a/swarms/structs/async_workflow.py b/swarms/structs/async_workflow.py index f0b8ac1e..56baecd3 100644 --- a/swarms/structs/async_workflow.py +++ b/swarms/structs/async_workflow.py @@ -2,10 +2,25 @@ import asyncio from typing import Any, List from swarms.structs.base_workflow import BaseWorkflow from swarms.structs.agent import Agent -from swarms.utils.loguru_logger import logger +from swarms.utils.loguru_logger import initialize_logger +logger = initialize_logger("async_workflow") class AsyncWorkflow(BaseWorkflow): + """ + Represents an asynchronous workflow that can execute tasks concurrently using multiple agents. + + Attributes: + - name (str): The name of the workflow. + - agents (List[Agent]): A list of agents participating in the workflow. + - max_workers (int): The maximum number of workers to use for concurrent execution. + - dashboard (bool): Indicates if a dashboard should be displayed. + - autosave (bool): Indicates if the results should be autosaved. + - verbose (bool): Indicates if verbose logging is enabled. + - task_pool (List): A pool of tasks to be executed. + - results (List): The results of the executed tasks. + - loop (asyncio.AbstractEventLoop): The event loop used for asynchronous execution. + """ def __init__( self, name: str = "AsyncWorkflow", @@ -30,7 +45,16 @@ class AsyncWorkflow(BaseWorkflow): async def _execute_agent_task( self, agent: Agent, task: str ) -> Any: - """Execute a single agent task asynchronously""" + """ + Executes a single agent task asynchronously. + + Args: + - agent (Agent): The agent executing the task. + - task (str): The task to be executed. + + Returns: + - Any: The result of the task execution or an error message if an exception occurs. + """ try: if self.verbose: logger.info( @@ -49,7 +73,15 @@ class AsyncWorkflow(BaseWorkflow): return str(e) async def run(self, task: str) -> List[Any]: - """Run the workflow with all agents processing the task concurrently""" + """ + Runs the workflow with all agents processing the task concurrently. + + Args: + - task (str): The task to be executed by all agents. + + Returns: + - List[Any]: A list of results from all agents or error messages if exceptions occur. + """ if not self.agents: raise ValueError("No agents provided to the workflow")