@ -15,34 +15,6 @@ The `ConcurrentWorkflow` class is designed to facilitate the concurrent executio
## Class Definitions
## Class Definitions
### AgentOutputSchema
The `AgentOutputSchema` class is a data model that captures the output and metadata for each agent's execution. It inherits from `pydantic.BaseModel` and provides structured fields to store essential information.
| `run_id` | `Optional[str]`| Unique ID for the run, automatically generated using `uuid`. |
| `agent_name` | `Optional[str]`| Name of the agent that executed the task. |
| `task` | `Optional[str]`| The task or query given to the agent. |
| `output` | `Optional[str]`| The output generated by the agent. |
| `start_time` | `Optional[datetime]`| The time when the agent started the task. |
| `end_time` | `Optional[datetime]`| The time when the agent completed the task. |
| `duration` | `Optional[float]` | The total time taken to complete the task, in seconds. |
### MetadataSchema
The `MetadataSchema` class is another data model that aggregates the outputs from all agents involved in the workflow. It also inherits from `pydantic.BaseModel` and includes fields for additional workflow-level metadata.
| `swarm_id` | `Optional[str]` | Unique ID for the workflow run, generated using `uuid`. |
| `task` | `Optional[str]` | The task or query given to all agents. |
| `description` | `Optional[str]` | A description of the workflow, typically indicating concurrent execution. |
| `agents` | `Optional[List[AgentOutputSchema]]` | A list of agent outputs and metadata. |
| `timestamp` | `Optional[datetime]` | The timestamp when the workflow was executed. |
## ConcurrentWorkflow
The `ConcurrentWorkflow` class is the core class that manages the concurrent execution of agents. It inherits from `BaseSwarm` and includes several key attributes and methods to facilitate this process.
The `ConcurrentWorkflow` class is the core class that manages the concurrent execution of agents. It inherits from `BaseSwarm` and includes several key attributes and methods to facilitate this process.
### Attributes
### Attributes
@ -59,6 +31,10 @@ The `ConcurrentWorkflow` class is the core class that manages the concurrent exe
| `return_str_on` | `bool` | Flag to return output as string. Defaults to `False`. |
| `return_str_on` | `bool` | Flag to return output as string. Defaults to `False`. |
| `agent_responses` | `List[str]` | List of agent responses as strings. |
| `agent_responses` | `List[str]` | List of agent responses as strings. |
| `auto_generate_prompts`| `bool` | Flag indicating whether to auto-generate prompts for agents. |
| `auto_generate_prompts`| `bool` | Flag indicating whether to auto-generate prompts for agents. |
| `output_type` | `OutputType` | Type of output format to return. Defaults to `"dict"`. |
| `return_entire_history`| `bool` | Flag to return entire conversation history. Defaults to `False`. |
| `agent` | `Agent` | The agent instance to run. |
| `task` | `str` | The task or query to give to the agent. |
| `executor` | `ThreadPoolExecutor` | The thread pool executor to use for running the agent task. |
#### Returns
- `AgentOutputSchema`: The metadata and output from the agent's execution.
#### Detailed Explanation
This method handles the execution of a single agent by offloading the task to a thread using `ThreadPoolExecutor`. It also tracks the time taken by the agent to complete the task and logs relevant information. If an exception occurs during execution, it captures the error and includes it in the output. The method implements retry logic for improved reliability.
Transforms the metadata schema into a string format.
Transforms the metadata schema into a string format.
@ -131,28 +89,6 @@ Transforms the metadata schema into a string format.
- `str`: The metadata schema as a formatted string.
- `str`: The metadata schema as a formatted string.
#### Detailed Explanation
This method converts the metadata stored in `MetadataSchema` into a human-readable string format, particularly focusing on the agent names and their respective outputs. This is useful for quickly reviewing the results of the concurrent workflow in a more accessible format.
| `task` | `str` | The task or query to give to all agents. |
#### Returns
- `MetadataSchema`: The aggregated metadata and outputs from all agents.
#### Detailed Explanation
This method is responsible for managing the concurrent execution of all agents. It uses `asyncio.gather` to run multiple agents simultaneously and collects their outputs into a `MetadataSchema` object. This aggregated metadata can then be saved or returned depending on the workflow configuration. The method includes retry logic for improved reliability.
### ConcurrentWorkflow.save_metadata
### ConcurrentWorkflow.save_metadata
Saves the metadata to a JSON file based on the `auto_save` flag.
Saves the metadata to a JSON file based on the `auto_save` flag.
@ -166,25 +102,29 @@ workflow.save_metadata()
### ConcurrentWorkflow.run
### ConcurrentWorkflow.run
Runs the workflow for the provided task, executes agents concurrently, and saves metadata.
- `Union[Dict[str, Any], str]`: The final metadata as a dictionary or a string, depending on the `return_str_on` flag.
- `Any`: The result of the execution, format depends on output_type and return_entire_history settings.
#### Detailed Explanation
#### Raises
This is the main method that a user will call to execute the workflow. It manages the entire process from starting the agents to collecting and optionally saving the metadata. The method also provides flexibility in how the results are returned—either as a JSON dictionary or as a formatted string.
- `ValueError`: If an invalid device is specified.
- `Exception`: If any other error occurs during execution.
### ConcurrentWorkflow.run_batched
### ConcurrentWorkflow.run_batched
Runs the workflow for a batch of tasks, executing agents concurrently for each task.
Runs the workflow for a batch of tasks.
#### Parameters
#### Parameters
@ -194,7 +134,7 @@ Runs the workflow for a batch of tasks, executing agents concurrently for each t
#### Returns
#### Returns
- `List[Union[Dict[str, Any], str]]`: A list of final metadata for each task, either as a dictionary or a string.
- `List[Union[Dict[str, Any], str]]`: A list of final metadata for each task.
| `tasks` | `List[str]` | A list of tasks or queries to give to all agents. |
#### Returns
- `List[asyncio.Future]`: A list of future objects representing the asynchronous operations for each task.
#### Example
```python
tasks = ["Task 1", "Task 2"]
futures = workflow.run_parallel_async(tasks)
results = await asyncio.gather(*futures)
print(results)
```
## Usage Examples
## Usage Examples
@ -303,15 +154,6 @@ print(results)
import os
import os
from swarms import Agent, ConcurrentWorkflow, OpenAIChat
from swarms import Agent, ConcurrentWorkflow, OpenAIChat
# Initialize agents
model = OpenAIChat(
api_key=os.getenv("OPENAI_API_KEY"),
model_name="gpt-4o-mini",
temperature=0.1,
)
# Define custom system prompts for each social media platform
# Define custom system prompts for each social media platform
TWITTER_AGENT_SYS_PROMPT = """
TWITTER_AGENT_SYS_PROMPT = """
You are a Twitter marketing expert specializing in real estate. Your task is to create engaging, concise tweets to promote properties, analyze trends to maximize engagement, and use appropriate hashtags and timing to reach potential buyers.
You are a Twitter marketing expert specializing in real estate. Your task is to create engaging, concise tweets to promote properties, analyze trends to maximize engagement, and use appropriate hashtags and timing to reach potential buyers.
description="The unique name assigned to the agent, which identifies its role and functionality within the swarm.",
)
description:Optional[str]=Field(
None,
description="A detailed explanation of the agent's purpose, capabilities, and any specific tasks it is designed to perform.",
)
system_prompt:Optional[str]=Field(
None,
description="The initial instruction or context provided to the agent, guiding its behavior and responses during execution.",
)
model_name:Optional[str]=Field(
description="The name of the AI model that the agent will utilize for processing tasks and generating outputs. For example: gpt-4o, gpt-4o-mini, openai/o3-mini"
)
auto_generate_prompt:Optional[bool]=Field(
description="A flag indicating whether the agent should automatically create prompts based on the task requirements."
)
max_tokens:Optional[int]=Field(
None,
description="The maximum number of tokens that the agent is allowed to generate in its responses, limiting output length.",
)
temperature:Optional[float]=Field(
description="A parameter that controls the randomness of the agent's output; lower values result in more deterministic responses."
)
role:Optional[str]=Field(
description="The designated role of the agent within the swarm, which influences its behavior and interaction with other agents."
)
max_loops:Optional[int]=Field(
description="The maximum number of times the agent is allowed to repeat its task, enabling iterative processing if necessary."
Kye Gomez
2 months ago