Kye Gomez
2 months ago
parent
fb494267eb
commit
928e1ecf45
@ -1,124 +0,0 @@
|
||||
# MultiProcessWorkflow Documentation
|
||||
|
||||
|
||||
The `MultiProcessWorkflow` class provides a framework for executing tasks concurrently using multiple processes. This class leverages Python's `multiprocessing` module to parallelize task execution, thereby enhancing performance and efficiency. It includes features such as automatic task retry on failure and optional autosaving of results. This documentation details the class, its parameters, attributes, methods, and usage examples.
|
||||
|
||||
## Class Definition
|
||||
|
||||
### `MultiProcessWorkflow`
|
||||
|
||||
|
||||
## Parameters
|
||||
|
||||
| Parameter | Type | Default | Description |
|
||||
|---------------|---------------------|---------|---------------------------------------------------------------|
|
||||
| `max_workers` | `int` | `5` | The maximum number of workers to use for parallel processing. |
|
||||
| `autosave` | `bool` | `True` | Flag indicating whether to automatically save the workflow. |
|
||||
| `agents` | `Sequence[Agent]` | `None` | A list of Agent objects representing the workflow agents. |
|
||||
| `*args` | `tuple` | | Additional positional arguments. |
|
||||
| `**kwargs` | `dict` | | Additional keyword arguments. |
|
||||
|
||||
## Attributes
|
||||
|
||||
| Attribute | Type | Description |
|
||||
|-----------------|---------------------|--------------------------------------------------------------|
|
||||
| `max_workers` | `int` | The maximum number of workers to use for parallel processing.|
|
||||
| `autosave` | `bool` | Flag indicating whether to automatically save the workflow. |
|
||||
| `agents` | `Sequence[Agent]` | A list of Agent objects representing the workflow agents. |
|
||||
|
||||
## Methods
|
||||
|
||||
### `execute_task`
|
||||
|
||||
#### Description
|
||||
|
||||
The `execute_task` method executes a given task and handles any exceptions that may occur during execution. If agents are defined, it will execute the task using each agent in sequence.
|
||||
|
||||
#### Usage Example
|
||||
|
||||
```python
|
||||
# Define a task
|
||||
task = Task()
|
||||
|
||||
# Execute the task
|
||||
workflow = MultiProcessWorkflow()
|
||||
result = workflow.execute_task(task)
|
||||
print(result)
|
||||
```
|
||||
|
||||
### `run`
|
||||
|
||||
#### Description
|
||||
|
||||
The `run` method executes the workflow by running the given task using multiple processes. It manages the task execution using a process pool and collects the results.
|
||||
|
||||
#### Usage Example
|
||||
|
||||
```python
|
||||
from swarms.structs.multi_process_workflow import MultiProcessingWorkflow
|
||||
from swarms.structs.task import Task
|
||||
from datetime import datetime
|
||||
from time import sleep
|
||||
|
||||
# Define a simple task
|
||||
def simple_task():
|
||||
sleep(1)
|
||||
return datetime.now()
|
||||
|
||||
# Create a task object
|
||||
task = Task(
|
||||
name="Simple Task",
|
||||
execute=simple_task,
|
||||
priority=1,
|
||||
)
|
||||
|
||||
# Create a workflow with the task
|
||||
workflow = MultiProcessWorkflow(max_workers=3, autosave=True, agents=[agent1, agent2])
|
||||
|
||||
# Run the workflow
|
||||
results = workflow.run(task)
|
||||
|
||||
# Print the results
|
||||
print(results)
|
||||
```
|
||||
|
||||
## Detailed Functionality and Usage
|
||||
|
||||
### Initialization
|
||||
|
||||
When an instance of `MultiProcessWorkflow` is created, it initializes the following:
|
||||
|
||||
- **max_workers**: Sets the maximum number of processes that can run concurrently.
|
||||
- **autosave**: Determines if the workflow results should be saved automatically.
|
||||
- **agents**: Accepts a list of agents that will perform the tasks.
|
||||
|
||||
### Running Tasks
|
||||
|
||||
The `run` method performs the following steps:
|
||||
|
||||
1. **Initialize Results and Manager**: Creates a list to store results and a `Manager` to manage shared state between processes.
|
||||
2. **Initialize Process Pool**: Creates a pool of worker processes.
|
||||
3. **Submit Tasks**: Iterates over the agents, submitting tasks to the pool for execution and collecting the results.
|
||||
4. **Wait for Completion**: Waits for all tasks to complete and collects the results.
|
||||
5. **Return Results**: Returns the list of results from all executed tasks.
|
||||
|
||||
### Autosave Task Result
|
||||
|
||||
Although the autosave functionality is mentioned in the parameters, it is not explicitly defined in the given code. The implementation for autosaving should be added based on the specific requirements of the application.
|
||||
|
||||
## Additional Information and Tips
|
||||
|
||||
- **Process Safety**: The use of `Manager` ensures that the list of results is managed safely across multiple processes.
|
||||
- **Logging**: The class uses the `logger` module to log information about task execution, retries, and failures.
|
||||
- **Error Handling**: The retry mechanism in the `execute_task` method helps in handling transient errors by attempting to re-execute failed tasks.
|
||||
|
||||
## References and Resources
|
||||
|
||||
For more information on multiprocessing in Python, refer to the following resources:
|
||||
|
||||
- [Python Multiprocessing Documentation](https://docs.python.org/3/library/multiprocessing.html)
|
||||
- [Python Logging Documentation](https://docs.python.org/3/library/logging.html)
|
||||
|
||||
---
|
||||
|
||||
By following this detailed documentation, users can effectively understand and utilize the `MultiProcessWorkflow` class to execute tasks concurrently with multiple processes. The examples provided help in demonstrating the practical usage of the class.
|
||||
@ -1,204 +0,0 @@
|
||||
# MultiProcessWorkflow Documentation
|
||||
|
||||
The `MultiProcessWorkflow` class extends the `BaseWorkflow` to support parallel processing using multiple workers. This class is designed to efficiently execute tasks concurrently, leveraging the power of multi-processing to enhance performance and scalability.
|
||||
|
||||
### Key Concepts
|
||||
|
||||
- **Parallel Processing**: Utilizing multiple workers to execute tasks concurrently.
|
||||
- **Workflow Management**: Handling the execution of tasks in a structured workflow.
|
||||
- **Agents**: Entities responsible for executing tasks.
|
||||
|
||||
## Attributes
|
||||
|
||||
### Arguments
|
||||
|
||||
| Argument | Type | Default | Description |
|
||||
|--------------|---------------------|---------|-------------|
|
||||
| `max_workers`| `int` | `5` | The maximum number of workers to use for parallel processing. |
|
||||
| `autosave` | `bool` | `True` | Flag indicating whether to automatically save the workflow. |
|
||||
| `agents` | `Sequence[Agent]` | `None` | A list of agents participating in the workflow. |
|
||||
| `*args` | | | Additional positional arguments. |
|
||||
| `**kwargs` | | | Additional keyword arguments. |
|
||||
|
||||
### Attributes
|
||||
|
||||
| Attribute | Type | Description |
|
||||
|--------------|---------------------|-------------|
|
||||
| `max_workers`| `int` | The maximum number of workers to use for parallel processing. |
|
||||
| `autosave` | `bool` | Flag indicating whether to automatically save the workflow. |
|
||||
| `agents` | `Sequence[Agent]` | A list of agents participating in the workflow. |
|
||||
|
||||
## Methods
|
||||
|
||||
### __init__
|
||||
|
||||
Initializes the `MultiProcessWorkflow` with the given parameters.
|
||||
|
||||
**Examples:**
|
||||
|
||||
```python
|
||||
from swarms.structs.agent import Agent
|
||||
from swarms.structs.task import Task
|
||||
from swarms.structs.multi_process_workflow import MultiProcessWorkflow
|
||||
|
||||
agents = [Agent(name="Agent 1"), Agent(name="Agent 2")]
|
||||
tasks = [Task(name="Task 1", execute=lambda: "result1"), Task(name="Task 2", execute=lambda: "result2")]
|
||||
|
||||
workflow = MultiProcessWorkflow(max_workers=3, agents=agents, tasks=tasks)
|
||||
```
|
||||
|
||||
### execute_task
|
||||
|
||||
Executes a task and handles exceptions.
|
||||
|
||||
**Arguments:**
|
||||
|
||||
| Parameter | Type | Description |
|
||||
|-----------|------|-------------|
|
||||
| `task` | `str` | The task to execute. |
|
||||
| `*args` | | Additional positional arguments for the task execution. |
|
||||
| `**kwargs`| | Additional keyword arguments for the task execution. |
|
||||
|
||||
**Returns:**
|
||||
|
||||
| Return Type | Description |
|
||||
|-------------|-------------|
|
||||
| `Any` | The result of the task execution. |
|
||||
|
||||
**Examples:**
|
||||
|
||||
```python
|
||||
result = workflow.execute_task(task="Sample Task")
|
||||
print(result)
|
||||
```
|
||||
|
||||
### run
|
||||
|
||||
Runs the workflow.
|
||||
|
||||
**Arguments:**
|
||||
|
||||
| Parameter | Type | Description |
|
||||
|-----------|------|-------------|
|
||||
| `task` | `str` | The task to run. |
|
||||
| `*args` | | Additional positional arguments for the task execution. |
|
||||
| `**kwargs`| | Additional keyword arguments for the task execution. |
|
||||
|
||||
**Returns:**
|
||||
|
||||
| Return Type | Description |
|
||||
|-------------|-------------|
|
||||
| `List[Any]` | The results of all executed tasks. |
|
||||
|
||||
**Examples:**
|
||||
|
||||
```python
|
||||
results = workflow.run(task="Sample Task")
|
||||
print(results)
|
||||
```
|
||||
|
||||
### Additional Examples
|
||||
|
||||
#### Example 1: Simple Task Execution
|
||||
|
||||
```python
|
||||
from swarms import Agent, Task, MultiProcessWorkflow, OpenAIChat
|
||||
from datetime import datetime
|
||||
from time import sleep
|
||||
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Load the environment variables
|
||||
load_dotenv()
|
||||
|
||||
|
||||
# Define a function to be used as the action
|
||||
def my_action():
|
||||
print("Action executed")
|
||||
|
||||
|
||||
# Define a function to be used as the condition
|
||||
def my_condition():
|
||||
print("Condition checked")
|
||||
return True
|
||||
|
||||
|
||||
# Create an agent
|
||||
agent = Agent(
|
||||
llm=OpenAIChat(openai_api_key=os.environ["OPENAI_API_KEY"]),
|
||||
max_loops=1,
|
||||
dashboard=False,
|
||||
)
|
||||
|
||||
# Create a task
|
||||
task = Task(
|
||||
description=(
|
||||
"Generate a report on the top 3 biggest expenses for small"
|
||||
" businesses and how businesses can save 20%"
|
||||
),
|
||||
agent=agent,
|
||||
)
|
||||
|
||||
# Create a workflow with the task
|
||||
workflow = MultiProcessWorkflow(tasks=[task])
|
||||
|
||||
# Run the workflow
|
||||
results = workflow.run(task)
|
||||
print(results)
|
||||
```
|
||||
|
||||
#### Example 2: Workflow with Multiple Agents
|
||||
|
||||
```python
|
||||
from swarms import Agent, Task, MultiProcessWorkflow
|
||||
|
||||
# Define tasks
|
||||
def task1():
|
||||
return "Task 1 result"
|
||||
|
||||
def task2():
|
||||
return "Task 2 result"
|
||||
|
||||
# Create agents
|
||||
agent1 = Agent(name="Agent 1", llm=OpenAIChat())
|
||||
agent2 = Agent(name="Agent 2", llm=OpenAIChat())
|
||||
|
||||
# Create tasks
|
||||
task_1 = Task(name="Task 1", execute=task1)
|
||||
task_2 = Task(name="Task 2", execute=task2)
|
||||
|
||||
# Create a workflow
|
||||
workflow = MultiProcessWorkflow(agents=[agent1, agent2], tasks=[task_1, task_2])
|
||||
|
||||
# Run the workflow
|
||||
results = workflow.run(task="Example Task")
|
||||
print(results)
|
||||
```
|
||||
|
||||
#### Example 3: Customizing Max Workers
|
||||
|
||||
```python
|
||||
from swarms import Agent, Task, MultiProcessWorkflow, OpenAIChat
|
||||
|
||||
# Define a task
|
||||
def example_task():
|
||||
return "Task result"
|
||||
|
||||
# Create an agent
|
||||
agent = Agent(name="Agent 1", llm=OpenAIChat())
|
||||
|
||||
# Create a task
|
||||
task = Task(name="Example Task", execute=example_task)
|
||||
|
||||
# Create a workflow with custom max workers
|
||||
workflow = MultiProcessWorkflow(max_workers=10, agents=[agent], tasks=[task])
|
||||
|
||||
# Run the workflow
|
||||
results = workflow.run(task="Example Task")
|
||||
print(results)
|
||||
```
|
||||
|
||||
## Summary
|
||||
|
||||
The `MultiProcessWorkflow` class provides a powerful framework for managing and executing tasks using multiple workers. With support for parallel processing, customizable workflows, and detailed logging, it is an ideal tool for complex task execution scenarios. This class enhances performance and scalability, making it suitable for a wide range of applications that require efficient task management.
|
||||
@ -0,0 +1,388 @@
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from typing import Tuple
|
||||
from litellm import completion
|
||||
from loguru import logger
|
||||
from swarms import Agent
|
||||
|
||||
EXTRACTION_PROMPT = """
|
||||
You are a specialized Chart2Table extraction agent that converts visual charts into precise textual descriptions.
|
||||
|
||||
Output Format:
|
||||
[Chart Type]
|
||||
Type: {bar|line|pie|scatter|combination}
|
||||
Title: {chart title}
|
||||
X-Axis: {label and scale}
|
||||
Y-Axis: {label and scale}
|
||||
|
||||
[Data Series]
|
||||
Name: {series name}
|
||||
Values: {comma-separated list of values}
|
||||
{repeat for each series}
|
||||
|
||||
[Annotations]
|
||||
- {list any markers, gridlines, legends}
|
||||
- {note any data gaps or anomalies}
|
||||
|
||||
Guidelines:
|
||||
1. Maintain exact numerical precision
|
||||
2. List ALL data points in order
|
||||
3. Note any gaps, outliers or special patterns
|
||||
4. Describe axes scales (linear/log) and units
|
||||
5. Include legends and series names verbatim
|
||||
6. Note any data point annotations or markers
|
||||
7. Describe chart elements spatially (top-left, center, etc)
|
||||
8. Include color and style information if relevant
|
||||
9. Note relationships between multiple series
|
||||
10. Flag any data quality or readability issues"""
|
||||
|
||||
REFORMULATION_PROMPT = """You are an Answer Reformulation specialist that breaks down complex analytical statements into atomic, verifiable claims.
|
||||
|
||||
Output Format:
|
||||
[Core Claims]
|
||||
1. {single fact with exact numbers}
|
||||
2. {another atomic fact}
|
||||
{continue for all core claims}
|
||||
|
||||
[Supporting Context]
|
||||
1. {relevant context that supports core claims}
|
||||
2. {additional contextual information}
|
||||
{continue for all context}
|
||||
|
||||
[Assumptions]
|
||||
1. {implicit assumption made}
|
||||
2. {another assumption}
|
||||
{continue for all assumptions}
|
||||
|
||||
Guidelines:
|
||||
1. Each claim must be independently verifiable
|
||||
2. Use exact numbers, never round or approximate
|
||||
3. Split compound statements into atomic facts
|
||||
4. Make implicit comparisons explicit
|
||||
5. Note temporal relationships clearly
|
||||
6. Include units with all measurements
|
||||
7. Flag any uncertainty or approximations
|
||||
8. Note data source limitations
|
||||
9. Preserve calculation steps
|
||||
10. Maintain logical dependencies"""
|
||||
|
||||
CAPTIONING_PROMPT = """You are an Entity Captioning specialist that generates rich contextual descriptions of chart elements.
|
||||
|
||||
Output Format:
|
||||
[Data Points]
|
||||
{x,y}: {detailed description of point significance}
|
||||
{continue for key points}
|
||||
|
||||
[Trends]
|
||||
- {description of overall pattern}
|
||||
- {notable sub-patterns}
|
||||
{continue for all trends}
|
||||
|
||||
[Relationships]
|
||||
- {correlation between variables}
|
||||
- {causation if evident}
|
||||
{continue for all relationships}
|
||||
|
||||
[Context]
|
||||
- {broader context for interpretation}
|
||||
- {relevant external factors}
|
||||
{continue for all context}
|
||||
|
||||
Guidelines:
|
||||
1. Describe both local and global patterns
|
||||
2. Note statistical significance of changes
|
||||
3. Identify cyclic or seasonal patterns
|
||||
4. Flag outliers and anomalies
|
||||
5. Compare relative magnitudes
|
||||
6. Note rate of change patterns
|
||||
7. Describe distribution characteristics
|
||||
8. Highlight key inflection points
|
||||
9. Note data clustering patterns
|
||||
10. Include domain-specific insights"""
|
||||
|
||||
PREFILTER_PROMPT = """You are a Pre-filtering specialist that identifies relevant chart elements for verification.
|
||||
|
||||
Output Format:
|
||||
[Critical Elements]
|
||||
1. {element}: Score {0-10}
|
||||
Evidence: {why this supports claims}
|
||||
{continue for all relevant elements}
|
||||
|
||||
[Supporting Elements]
|
||||
1. {element}: Score {0-10}
|
||||
Context: {how this adds context}
|
||||
{continue for all supporting elements}
|
||||
|
||||
[Relevance Chain]
|
||||
1. {claim} -> {element} -> {evidence}
|
||||
{continue for all connections}
|
||||
|
||||
Guidelines:
|
||||
1. Score relevance 0-10 with detailed rationale
|
||||
2. Build explicit evidence chains
|
||||
3. Note both direct and indirect support
|
||||
4. Consider temporal relevance
|
||||
5. Account for data relationships
|
||||
6. Note confidence levels
|
||||
7. Include contextual importance
|
||||
8. Consider alternative interpretations
|
||||
9. Note missing evidence
|
||||
10. Explain filtering decisions"""
|
||||
|
||||
RERANK_PROMPT = """You are a Re-ranking specialist that orders evidence by strength and relevance.
|
||||
|
||||
Output Format:
|
||||
[Primary Evidence]
|
||||
1. {element} - Score: {0-10}
|
||||
Strength: {detailed justification}
|
||||
{continue for top evidence}
|
||||
|
||||
[Supporting Evidence]
|
||||
1. {element} - Score: {0-10}
|
||||
Context: {how this reinforces primary evidence}
|
||||
{continue for supporting evidence}
|
||||
|
||||
[Evidence Chains]
|
||||
1. {claim} -> {primary} -> {supporting} -> {conclusion}
|
||||
{continue for all chains}
|
||||
|
||||
Guidelines:
|
||||
1. Use explicit scoring criteria
|
||||
2. Consider evidence independence
|
||||
3. Note corroborating elements
|
||||
4. Account for evidence quality
|
||||
5. Consider contradictory evidence
|
||||
6. Note confidence levels
|
||||
7. Explain ranking decisions
|
||||
8. Build complete evidence chains
|
||||
9. Note gaps in evidence
|
||||
10. Consider alternative interpretations"""
|
||||
|
||||
LOCALIZATION_PROMPT = """You are a Cell Localization specialist that precisely maps data to visual elements.
|
||||
|
||||
Output Format:
|
||||
[Element Locations]
|
||||
1. Type: {bar|line|point|label}
|
||||
Position: {x1,y1,x2,y2}
|
||||
Value: {associated data value}
|
||||
Confidence: {0-10}
|
||||
{continue for all elements}
|
||||
|
||||
[Spatial Relationships]
|
||||
- {relative positions}
|
||||
- {alignment patterns}
|
||||
{continue for all relationships}
|
||||
|
||||
[Visual Context]
|
||||
- {surrounding elements}
|
||||
- {reference points}
|
||||
{continue for context}
|
||||
|
||||
Guidelines:
|
||||
1. Use normalized coordinates (0-1)
|
||||
2. Note element boundaries precisely
|
||||
3. Include confidence scores
|
||||
4. Note spatial relationships
|
||||
5. Account for overlapping elements
|
||||
6. Consider chart type constraints
|
||||
7. Note alignment patterns
|
||||
8. Include reference points
|
||||
9. Note visual hierarchies
|
||||
10. Document occlusions"""
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChartElement:
|
||||
element_type: str
|
||||
bbox: Tuple[float, float, float, float]
|
||||
confidence: float
|
||||
|
||||
|
||||
class VisionAPI:
|
||||
def __init__(
|
||||
self,
|
||||
model_name: str = "gpt-4o",
|
||||
max_tokens: int = 1000,
|
||||
temperature: float = 0.5,
|
||||
):
|
||||
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY")
|
||||
self.model_name = model_name
|
||||
self.max_tokens = max_tokens
|
||||
self.temperature = temperature
|
||||
|
||||
def encode_image(self, img: str):
|
||||
if img.startswith("http"):
|
||||
return img
|
||||
import base64
|
||||
|
||||
with open(img, "rb") as image_file:
|
||||
encoded_string = base64.b64encode(
|
||||
image_file.read()
|
||||
).decode("utf-8")
|
||||
return f"data:image/png;base64,{encoded_string}"
|
||||
|
||||
def run(self, task: str, img: str):
|
||||
img = self.encode_image(img)
|
||||
response = completion(
|
||||
model=self.model_name,
|
||||
messages=[
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": task},
|
||||
{
|
||||
"type": "image_url",
|
||||
"image_url": {"url": img},
|
||||
},
|
||||
],
|
||||
}
|
||||
],
|
||||
max_tokens=self.max_tokens,
|
||||
temperature=self.temperature,
|
||||
)
|
||||
return response.choices[0].message.content
|
||||
|
||||
|
||||
class ChartCitor:
|
||||
def __init__(
|
||||
self,
|
||||
model_name: str = "gpt-4o",
|
||||
saved_state_path: str = "chartcitor_state.json",
|
||||
max_retries: int = 3,
|
||||
max_loops: int = 1,
|
||||
):
|
||||
logger.info(
|
||||
f"Initializing ChartCitor with model {model_name}"
|
||||
)
|
||||
model = VisionAPI()
|
||||
|
||||
self.extraction_agent = Agent(
|
||||
agent_name="Chart2Table-Agent",
|
||||
system_prompt=EXTRACTION_PROMPT,
|
||||
llm=model,
|
||||
max_loops=1,
|
||||
)
|
||||
|
||||
self.reformulation_agent = Agent(
|
||||
agent_name="Answer-Reformulation-Agent",
|
||||
system_prompt=REFORMULATION_PROMPT,
|
||||
llm=model,
|
||||
max_loops=1,
|
||||
)
|
||||
|
||||
self.captioning_agent = Agent(
|
||||
agent_name="Entity-Captioning-Agent",
|
||||
system_prompt=CAPTIONING_PROMPT,
|
||||
llm=model,
|
||||
max_loops=1,
|
||||
)
|
||||
|
||||
self.prefilter_agent = Agent(
|
||||
agent_name="LLM-Prefilter-Agent",
|
||||
system_prompt=PREFILTER_PROMPT,
|
||||
llm=model,
|
||||
max_loops=1,
|
||||
)
|
||||
|
||||
self.rerank_agent = Agent(
|
||||
agent_name="LLM-Rerank-Agent",
|
||||
system_prompt=RERANK_PROMPT,
|
||||
llm=model,
|
||||
max_loops=1,
|
||||
)
|
||||
|
||||
self.localization_agent = Agent(
|
||||
agent_name="Cell-Localization-Agent",
|
||||
system_prompt=LOCALIZATION_PROMPT,
|
||||
llm=model,
|
||||
max_loops=1,
|
||||
)
|
||||
|
||||
def extract_table(self, chart_image: str) -> str:
|
||||
logger.info("Extracting table from chart")
|
||||
return self.extraction_agent.run(
|
||||
"Extract and describe the data from this chart following the specified format.",
|
||||
img=chart_image,
|
||||
)
|
||||
|
||||
def reformulate_answer(
|
||||
self, answer: str, table_data: str, chart_image: str
|
||||
) -> str:
|
||||
logger.info("Reformulating answer into atomic facts")
|
||||
return self.reformulation_agent.run(
|
||||
f"Break this answer into atomic facts:\n{answer}\n\nTable data:\n{table_data}",
|
||||
img=chart_image,
|
||||
)
|
||||
|
||||
def generate_captions(
|
||||
self, table_data: str, chart_image: str
|
||||
) -> str:
|
||||
logger.info("Generating captions for chart elements")
|
||||
return self.captioning_agent.run(
|
||||
f"Generate descriptive captions for this data:\n{table_data}",
|
||||
img=chart_image,
|
||||
)
|
||||
|
||||
def retrieve_evidence(
|
||||
self,
|
||||
facts: str,
|
||||
table_data: str,
|
||||
captions: str,
|
||||
chart_image: str,
|
||||
) -> str:
|
||||
logger.info("Retrieving supporting evidence")
|
||||
filtered = self.prefilter_agent.run(
|
||||
f"Identify relevant elements for:\nFacts:\n{facts}\n\nData:\n{table_data}\n\nCaptions:\n{captions}",
|
||||
img=chart_image,
|
||||
)
|
||||
|
||||
return self.rerank_agent.run(
|
||||
f"Rank these elements by relevance:\n{filtered}\nFor facts:\n{facts}",
|
||||
img=chart_image,
|
||||
)
|
||||
|
||||
def localize_elements(
|
||||
self, chart_image: str, evidence: str
|
||||
) -> str:
|
||||
logger.info("Localizing chart elements")
|
||||
return self.localization_agent.run(
|
||||
f"Describe the location of these elements:\n{evidence}",
|
||||
img=chart_image,
|
||||
)
|
||||
|
||||
def run(
|
||||
self, chart_image: str, question: str, answer: str
|
||||
) -> str:
|
||||
logger.info(f"Processing question: {question}")
|
||||
|
||||
table_data = self.extract_table(chart_image)
|
||||
facts = self.reformulate_answer(
|
||||
answer, table_data, chart_image
|
||||
)
|
||||
captions = self.generate_captions(table_data, chart_image)
|
||||
evidence = self.retrieve_evidence(
|
||||
facts, table_data, captions, chart_image
|
||||
)
|
||||
citations = self.localize_elements(chart_image, evidence)
|
||||
|
||||
return f"""Analysis Results:
|
||||
|
||||
Facts:
|
||||
{facts}
|
||||
|
||||
Evidence:
|
||||
{evidence}
|
||||
|
||||
Visual Citations:
|
||||
{citations}
|
||||
"""
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
chartcitor = ChartCitor()
|
||||
result = chartcitor.run(
|
||||
chart_image="chart.png",
|
||||
question="Analyze this chart of solana price and volume over time. What is the highest volume day?",
|
||||
answer="203",
|
||||
)
|
||||
print(result)
|
||||
@ -0,0 +1,156 @@
|
||||
import os
|
||||
from swarms import Agent
|
||||
from dotenv import load_dotenv
|
||||
|
||||
from swarm_models import OpenAIChat
|
||||
|
||||
load_dotenv()
|
||||
|
||||
|
||||
model = OpenAIChat(
|
||||
model_name="deepseek-ai/DeepSeek-R1-Distill-Llama-70B-free",
|
||||
openai_api_key=os.getenv("TOGETHER_API_KEY"),
|
||||
base_url="https://api.together.xyz/v1",
|
||||
)
|
||||
|
||||
# Define system prompts for reasoning agents
|
||||
THINKING_AGENT_PROMPT = """You are a sophisticated analytical and strategic thinking agent focused on deep problem analysis and solution design.
|
||||
|
||||
Your core capabilities include:
|
||||
1. Comprehensive Problem Analysis
|
||||
- Break down complex problems into constituent elements
|
||||
- Map relationships and dependencies between components
|
||||
- Identify root causes and underlying patterns
|
||||
- Consider historical context and precedents
|
||||
|
||||
2. Multi-Perspective Evaluation
|
||||
- Examine issues from multiple stakeholder viewpoints
|
||||
- Consider short-term and long-term implications
|
||||
- Evaluate social, economic, technical, and ethical dimensions
|
||||
- Challenge assumptions and identify potential biases
|
||||
|
||||
3. Risk Assessment and Mitigation
|
||||
- Conduct thorough risk analysis across scenarios
|
||||
- Identify potential failure modes and edge cases
|
||||
- Develop contingency plans and mitigation strategies
|
||||
- Assess probability and impact of various outcomes
|
||||
|
||||
4. Strategic Solution Development
|
||||
- Generate multiple solution approaches
|
||||
- Evaluate trade-offs between different strategies
|
||||
- Consider resource constraints and limitations
|
||||
- Design scalable and sustainable solutions
|
||||
|
||||
5. Decision Framework Creation
|
||||
- Establish clear evaluation criteria
|
||||
- Weight competing priorities appropriately
|
||||
- Create structured decision matrices
|
||||
- Document reasoning and key decision factors
|
||||
|
||||
6. Systems Thinking
|
||||
- Map interconnections between system elements
|
||||
- Identify feedback loops and cascade effects
|
||||
- Consider emergent properties and behaviors
|
||||
- Account for dynamic system evolution
|
||||
|
||||
Your output should always include:
|
||||
- Clear articulation of your analytical process
|
||||
- Key assumptions and their justification
|
||||
- Potential risks and mitigation strategies
|
||||
- Multiple solution options with pros/cons
|
||||
- Specific recommendations with supporting rationale
|
||||
- Areas of uncertainty requiring further investigation
|
||||
|
||||
Focus on developing robust, well-reasoned strategies that account for complexity while remaining practical and actionable."""
|
||||
|
||||
ACTION_AGENT_PROMPT = """You are an advanced implementation and execution agent focused on turning strategic plans into concrete results.
|
||||
|
||||
Your core capabilities include:
|
||||
1. Strategic Implementation Planning
|
||||
- Break down high-level strategies into specific actions
|
||||
- Create detailed project roadmaps and timelines
|
||||
- Identify critical path dependencies
|
||||
- Establish clear milestones and success metrics
|
||||
- Design feedback and monitoring mechanisms
|
||||
|
||||
2. Resource Optimization
|
||||
- Assess resource requirements and constraints
|
||||
- Optimize resource allocation and scheduling
|
||||
- Identify efficiency opportunities
|
||||
- Plan for scalability and flexibility
|
||||
- Manage competing priorities effectively
|
||||
|
||||
3. Execution Management
|
||||
- Develop detailed implementation procedures
|
||||
- Create clear operational guidelines
|
||||
- Establish quality control measures
|
||||
- Design progress tracking systems
|
||||
- Build in review and adjustment points
|
||||
|
||||
4. Risk Management
|
||||
- Implement specific risk mitigation measures
|
||||
- Create early warning systems
|
||||
- Develop contingency procedures
|
||||
- Establish fallback positions
|
||||
- Monitor risk indicators
|
||||
|
||||
5. Stakeholder Management
|
||||
- Identify key stakeholders and their needs
|
||||
- Create communication plans
|
||||
- Establish feedback mechanisms
|
||||
- Manage expectations effectively
|
||||
- Build support and buy-in
|
||||
|
||||
6. Continuous Improvement
|
||||
- Monitor implementation effectiveness
|
||||
- Gather and analyze performance data
|
||||
- Identify improvement opportunities
|
||||
- Implement iterative enhancements
|
||||
- Document lessons learned
|
||||
|
||||
Your output should always include:
|
||||
- Detailed action plans with specific steps
|
||||
- Resource requirements and allocation plans
|
||||
- Timeline with key milestones
|
||||
- Success metrics and monitoring approach
|
||||
- Risk mitigation procedures
|
||||
- Communication and stakeholder management plans
|
||||
- Quality control measures
|
||||
- Feedback and adjustment mechanisms
|
||||
|
||||
Focus on practical, efficient, and effective implementation while maintaining high quality standards and achieving desired outcomes."""
|
||||
|
||||
# Initialize the thinking agent
|
||||
thinking_agent = Agent(
|
||||
agent_name="Strategic-Thinker",
|
||||
agent_description="Deep analysis and strategic planning agent",
|
||||
system_prompt=THINKING_AGENT_PROMPT,
|
||||
max_loops=1,
|
||||
llm=model,
|
||||
dynamic_temperature_enabled=True,
|
||||
)
|
||||
|
||||
# Initialize the action agent
|
||||
action_agent = Agent(
|
||||
agent_name="Action-Executor",
|
||||
agent_description="Practical implementation and execution agent",
|
||||
system_prompt=ACTION_AGENT_PROMPT,
|
||||
max_loops=1,
|
||||
model_name="gpt-4o",
|
||||
dynamic_temperature_enabled=True,
|
||||
)
|
||||
|
||||
|
||||
def run_reasoning_duo(task: str):
|
||||
# Step 1: Thinking Agent
|
||||
thinking_result = thinking_agent.run(task)
|
||||
|
||||
# Step 2: Action Agent
|
||||
action_result = action_agent.run(
|
||||
f"From {thinking_agent.agent_name}: {thinking_result}"
|
||||
)
|
||||
return action_result
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
run_reasoning_duo("What is the best way to invest $1000?")
|
||||
@ -0,0 +1,170 @@
|
||||
from loguru import logger
|
||||
from swarms.structs.swarm_eval import (
|
||||
SwarmEvaluator,
|
||||
PRESET_DATASETS,
|
||||
)
|
||||
|
||||
import os
|
||||
from swarms import Agent
|
||||
from dotenv import load_dotenv
|
||||
|
||||
from swarm_models import OpenAIChat
|
||||
|
||||
load_dotenv()
|
||||
|
||||
|
||||
model = OpenAIChat(
|
||||
model_name="deepseek-ai/DeepSeek-R1-Distill-Llama-70B-free",
|
||||
openai_api_key=os.getenv("TOGETHER_API_KEY"),
|
||||
base_url="https://api.together.xyz/v1",
|
||||
)
|
||||
|
||||
# Define system prompts for reasoning agents
|
||||
THINKING_AGENT_PROMPT = """You are a sophisticated analytical and strategic thinking agent focused on deep problem analysis and solution design.
|
||||
|
||||
Your core capabilities include:
|
||||
1. Comprehensive Problem Analysis
|
||||
- Break down complex problems into constituent elements
|
||||
- Map relationships and dependencies between components
|
||||
- Identify root causes and underlying patterns
|
||||
- Consider historical context and precedents
|
||||
|
||||
2. Multi-Perspective Evaluation
|
||||
- Examine issues from multiple stakeholder viewpoints
|
||||
- Consider short-term and long-term implications
|
||||
- Evaluate social, economic, technical, and ethical dimensions
|
||||
- Challenge assumptions and identify potential biases
|
||||
|
||||
3. Risk Assessment and Mitigation
|
||||
- Conduct thorough risk analysis across scenarios
|
||||
- Identify potential failure modes and edge cases
|
||||
- Develop contingency plans and mitigation strategies
|
||||
- Assess probability and impact of various outcomes
|
||||
|
||||
4. Strategic Solution Development
|
||||
- Generate multiple solution approaches
|
||||
- Evaluate trade-offs between different strategies
|
||||
- Consider resource constraints and limitations
|
||||
- Design scalable and sustainable solutions
|
||||
|
||||
5. Decision Framework Creation
|
||||
- Establish clear evaluation criteria
|
||||
- Weight competing priorities appropriately
|
||||
- Create structured decision matrices
|
||||
- Document reasoning and key decision factors
|
||||
|
||||
6. Systems Thinking
|
||||
- Map interconnections between system elements
|
||||
- Identify feedback loops and cascade effects
|
||||
- Consider emergent properties and behaviors
|
||||
- Account for dynamic system evolution
|
||||
|
||||
Your output should always include:
|
||||
- Clear articulation of your analytical process
|
||||
- Key assumptions and their justification
|
||||
- Potential risks and mitigation strategies
|
||||
- Multiple solution options with pros/cons
|
||||
- Specific recommendations with supporting rationale
|
||||
- Areas of uncertainty requiring further investigation
|
||||
|
||||
Focus on developing robust, well-reasoned strategies that account for complexity while remaining practical and actionable."""
|
||||
|
||||
ACTION_AGENT_PROMPT = """You are an advanced implementation and execution agent focused on turning strategic plans into concrete results.
|
||||
|
||||
Your core capabilities include:
|
||||
1. Strategic Implementation Planning
|
||||
- Break down high-level strategies into specific actions
|
||||
- Create detailed project roadmaps and timelines
|
||||
- Identify critical path dependencies
|
||||
- Establish clear milestones and success metrics
|
||||
- Design feedback and monitoring mechanisms
|
||||
|
||||
2. Resource Optimization
|
||||
- Assess resource requirements and constraints
|
||||
- Optimize resource allocation and scheduling
|
||||
- Identify efficiency opportunities
|
||||
- Plan for scalability and flexibility
|
||||
- Manage competing priorities effectively
|
||||
|
||||
3. Execution Management
|
||||
- Develop detailed implementation procedures
|
||||
- Create clear operational guidelines
|
||||
- Establish quality control measures
|
||||
- Design progress tracking systems
|
||||
- Build in review and adjustment points
|
||||
|
||||
4. Risk Management
|
||||
- Implement specific risk mitigation measures
|
||||
- Create early warning systems
|
||||
- Develop contingency procedures
|
||||
- Establish fallback positions
|
||||
- Monitor risk indicators
|
||||
|
||||
5. Stakeholder Management
|
||||
- Identify key stakeholders and their needs
|
||||
- Create communication plans
|
||||
- Establish feedback mechanisms
|
||||
- Manage expectations effectively
|
||||
- Build support and buy-in
|
||||
|
||||
6. Continuous Improvement
|
||||
- Monitor implementation effectiveness
|
||||
- Gather and analyze performance data
|
||||
- Identify improvement opportunities
|
||||
- Implement iterative enhancements
|
||||
- Document lessons learned
|
||||
|
||||
Your output should always include:
|
||||
- Detailed action plans with specific steps
|
||||
- Resource requirements and allocation plans
|
||||
- Timeline with key milestones
|
||||
- Success metrics and monitoring approach
|
||||
- Risk mitigation procedures
|
||||
- Communication and stakeholder management plans
|
||||
- Quality control measures
|
||||
- Feedback and adjustment mechanisms
|
||||
|
||||
Focus on practical, efficient, and effective implementation while maintaining high quality standards and achieving desired outcomes."""
|
||||
|
||||
# Initialize the thinking agent
|
||||
thinking_agent = Agent(
|
||||
agent_name="Strategic-Thinker",
|
||||
agent_description="Deep analysis and strategic planning agent",
|
||||
system_prompt=THINKING_AGENT_PROMPT,
|
||||
max_loops=1,
|
||||
llm=model,
|
||||
dynamic_temperature_enabled=True,
|
||||
)
|
||||
|
||||
|
||||
class DeepSeekSwarm:
|
||||
def __init__(self):
|
||||
self.thinking_agent = thinking_agent
|
||||
|
||||
def run(self, task: str):
|
||||
first_one = self.thinking_agent.run(task)
|
||||
|
||||
return self.thinking_agent.run(first_one)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Initialize the swarm (replace with your actual multi-agent system)
|
||||
swarm = DeepSeekSwarm()
|
||||
|
||||
# Initialize the evaluator with the swarm instance
|
||||
evaluator = SwarmEvaluator(swarm)
|
||||
|
||||
logger.info("Starting evaluation for dataset: gsm8k")
|
||||
|
||||
# For demonstration, we use 4 concurrent workers, show progress, and save results.
|
||||
results = evaluator.evaluate(
|
||||
"gsm8k",
|
||||
split="train",
|
||||
config=PRESET_DATASETS["gsm8k"],
|
||||
max_workers=os.cpu_count(),
|
||||
max_retries=3,
|
||||
show_progress=True,
|
||||
output_file="gsm8k_results.txt",
|
||||
)
|
||||
|
||||
logger.info(f"Results for gsm8k: {results}")
|
||||
@ -0,0 +1,52 @@
|
||||
from swarms import Agent, MajorityVoting
|
||||
|
||||
# Initialize multiple agents with different specialties
|
||||
agents = [
|
||||
Agent(
|
||||
agent_name="Financial-Analysis-Agent",
|
||||
agent_description="Personal finance advisor focused on market analysis",
|
||||
system_prompt="You are a financial advisor specializing in market analysis and investment opportunities.",
|
||||
max_loops=1,
|
||||
model_name="gpt-4o"
|
||||
),
|
||||
Agent(
|
||||
agent_name="Risk-Assessment-Agent",
|
||||
agent_description="Risk analysis and portfolio management expert",
|
||||
system_prompt="You are a risk assessment expert focused on evaluating investment risks and portfolio diversification.",
|
||||
max_loops=1,
|
||||
model_name="gpt-4o"
|
||||
),
|
||||
Agent(
|
||||
agent_name="Tech-Investment-Agent",
|
||||
agent_description="Technology sector investment specialist",
|
||||
system_prompt="You are a technology investment specialist focused on AI, emerging tech, and growth opportunities.",
|
||||
max_loops=1,
|
||||
model_name="gpt-4o"
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
consensus_agent = Agent(
|
||||
agent_name="Consensus-Agent",
|
||||
agent_description="Consensus agent focused on analyzing investment advice",
|
||||
system_prompt="You are a consensus agent focused on analyzing investment advice and providing a final answer.",
|
||||
max_loops=1,
|
||||
model_name="gpt-4o"
|
||||
)
|
||||
|
||||
# Create majority voting system
|
||||
majority_voting = MajorityVoting(
|
||||
name="Investment-Advisory-System",
|
||||
description="Multi-agent system for investment advice",
|
||||
agents=agents,
|
||||
verbose=True,
|
||||
consensus_agent=consensus_agent
|
||||
)
|
||||
|
||||
# Run the analysis with majority voting
|
||||
result = majority_voting.run(
|
||||
task="Create a table of super high growth opportunities for AI. I have $40k to invest in ETFs, index funds, and more. Please create a table in markdown.",
|
||||
correct_answer="" # Optional evaluation metric
|
||||
)
|
||||
|
||||
print(result)
|
||||
@ -0,0 +1,18 @@
|
||||
from swarms import Agent
|
||||
from swarms.prompts.finance_agent_sys_prompt import (
|
||||
FINANCIAL_AGENT_SYS_PROMPT,
|
||||
)
|
||||
|
||||
|
||||
# Initialize the agent
|
||||
agent = Agent(
|
||||
agent_name="Financial-Analysis-Agent",
|
||||
agent_description="Personal finance advisor agent",
|
||||
system_prompt=FINANCIAL_AGENT_SYS_PROMPT,
|
||||
max_loops=1,
|
||||
model_name="ollama/llama2",
|
||||
)
|
||||
|
||||
agent.run(
|
||||
"Create a table of super high growth opportunities for AI. I have $40k to invest in ETFs, index funds, and more. Please create a table in markdown.",
|
||||
)
|
||||
@ -1,244 +0,0 @@
|
||||
from multiprocessing import Manager, Pool, cpu_count
|
||||
from typing import Sequence, Union, Callable, List
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
from swarms.structs.agent import Agent
|
||||
from swarms.structs.base_workflow import BaseWorkflow
|
||||
from swarms.utils.loguru_logger import initialize_logger
|
||||
|
||||
logger = initialize_logger(log_folder="multi_process_workflow")
|
||||
|
||||
|
||||
class MultiProcessWorkflow(BaseWorkflow):
|
||||
"""
|
||||
Initialize a MultiProcessWorkflow object.
|
||||
|
||||
Args:
|
||||
max_workers (int): The maximum number of workers to use for parallel processing.
|
||||
autosave (bool): Flag indicating whether to automatically save the workflow.
|
||||
agents (List[Union[Agent, Callable]]): A list of Agent objects or callable functions representing the workflow tasks.
|
||||
*args: Additional positional arguments.
|
||||
**kwargs: Additional keyword arguments.
|
||||
|
||||
Example:
|
||||
>>> from swarms.structs.multi_process_workflow import MultiProcessingWorkflow
|
||||
>>> from swarms.structs.task import Task
|
||||
>>> from datetime import datetime
|
||||
>>> from time import sleep
|
||||
>>>
|
||||
>>> # Define a simple task
|
||||
>>> def simple_task():
|
||||
>>> sleep(1)
|
||||
>>> return datetime.now()
|
||||
>>>
|
||||
>>> # Create a task object
|
||||
>>> task = Task(
|
||||
>>> name="Simple Task",
|
||||
>>> execute=simple_task,
|
||||
>>> priority=1,
|
||||
>>> )
|
||||
>>>
|
||||
>>> # Create a workflow with the task
|
||||
>>> workflow = MultiProcessingWorkflow(tasks=[task])
|
||||
>>>
|
||||
>>> # Run the workflow
|
||||
>>> results = workflow.run(task)
|
||||
>>>
|
||||
>>> # Print the results
|
||||
>>> print(results)
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
max_workers: int = 5,
|
||||
autosave: bool = True,
|
||||
agents: Sequence[Union[Agent, Callable]] = None,
|
||||
*args,
|
||||
**kwargs,
|
||||
):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.max_workers = max_workers
|
||||
self.autosave = autosave
|
||||
self.agents = agents
|
||||
|
||||
self.max_workers or cpu_count()
|
||||
|
||||
# Log
|
||||
logger.info(
|
||||
(
|
||||
"Initialized MultiProcessWorkflow with"
|
||||
f" {self.max_workers} max workers and autosave set to"
|
||||
f" {self.autosave}"
|
||||
),
|
||||
)
|
||||
|
||||
# Log the agents
|
||||
if self.agents is not None:
|
||||
for agent in self.agents:
|
||||
logger.info(f"Agent: {agent.agent_name}")
|
||||
|
||||
def execute_task(self, task: str, *args, **kwargs):
|
||||
"""Execute a task and handle exceptions.
|
||||
|
||||
Args:
|
||||
task (Task): The task to execute.
|
||||
*args: Additional positional arguments for the task execution.
|
||||
**kwargs: Additional keyword arguments for the task execution.
|
||||
|
||||
Returns:
|
||||
Any: The result of the task execution.
|
||||
|
||||
"""
|
||||
try:
|
||||
if self.agents is not None:
|
||||
# Execute the task
|
||||
for agent in self.agents:
|
||||
result = agent.run(task, *args, **kwargs)
|
||||
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
(
|
||||
"An error occurred during execution of task"
|
||||
f" {task}: {str(e)}"
|
||||
),
|
||||
)
|
||||
return None
|
||||
|
||||
def run(self, task: str, *args, **kwargs):
|
||||
"""Run the workflow.
|
||||
|
||||
Args:
|
||||
task (Task): The task to run.
|
||||
*args: Additional positional arguments for the task execution.
|
||||
**kwargs: Additional keyword arguments for the task execution.
|
||||
|
||||
Returns:
|
||||
List[Any]: The results of all executed tasks.
|
||||
|
||||
"""
|
||||
try:
|
||||
results = []
|
||||
with Manager() as manager:
|
||||
with Pool(
|
||||
processes=self.max_workers, *args, **kwargs
|
||||
) as pool:
|
||||
# Using manager.list() to collect results in a process safe way
|
||||
results_list = manager.list()
|
||||
jobs = [
|
||||
pool.apply_async(
|
||||
self.execute_task, # Pass the function, not the function call
|
||||
args=(task,)
|
||||
+ args, # Pass the arguments as a tuple
|
||||
kwds=kwargs, # Pass the keyword arguments as a dictionary
|
||||
callback=results_list.append,
|
||||
timeout=task.timeout,
|
||||
)
|
||||
for agent in self.agents
|
||||
]
|
||||
|
||||
# Wait for all jobs to complete
|
||||
for job in jobs:
|
||||
job.get()
|
||||
|
||||
results = list(results_list)
|
||||
|
||||
return results
|
||||
except Exception as error:
|
||||
logger.error(f"Error in run: {error}")
|
||||
return None
|
||||
|
||||
async def async_run(self, task: str, *args, **kwargs):
|
||||
"""Asynchronously run the workflow.
|
||||
|
||||
Args:
|
||||
task (Task): The task to run.
|
||||
*args: Additional positional arguments for the task execution.
|
||||
**kwargs: Additional keyword arguments for the task execution.
|
||||
|
||||
Returns:
|
||||
List[Any]: The results of all executed tasks.
|
||||
|
||||
"""
|
||||
try:
|
||||
results = []
|
||||
with ThreadPoolExecutor(
|
||||
max_workers=self.max_workers
|
||||
) as executor:
|
||||
futures = [
|
||||
executor.submit(
|
||||
self.execute_task, task, *args, **kwargs
|
||||
)
|
||||
for _ in range(len(self.agents))
|
||||
]
|
||||
for future in as_completed(futures):
|
||||
result = future.result()
|
||||
results.append(result)
|
||||
|
||||
return results
|
||||
except Exception as error:
|
||||
logger.error(f"Error in async_run: {error}")
|
||||
return None
|
||||
|
||||
def batched_run(
|
||||
self, tasks: List[str], batch_size: int = 5, *args, **kwargs
|
||||
):
|
||||
"""Run tasks in batches.
|
||||
|
||||
Args:
|
||||
tasks (List[str]): A list of tasks to run.
|
||||
batch_size (int): The size of each batch.
|
||||
*args: Additional positional arguments for the task execution.
|
||||
**kwargs: Additional keyword arguments for the task execution.
|
||||
|
||||
Returns:
|
||||
List[Any]: The results of all executed tasks.
|
||||
|
||||
"""
|
||||
try:
|
||||
results = []
|
||||
for i in range(0, len(tasks), batch_size):
|
||||
batch = tasks[i : i + batch_size]
|
||||
with Pool(processes=self.max_workers) as pool:
|
||||
results_list = pool.map(
|
||||
self.execute_task, batch, *args, **kwargs
|
||||
)
|
||||
results.extend(results_list)
|
||||
|
||||
return results
|
||||
except Exception as error:
|
||||
logger.error(f"Error in batched_run: {error}")
|
||||
return None
|
||||
|
||||
def concurrent_run(self, tasks: List[str], *args, **kwargs):
|
||||
"""Run tasks concurrently.
|
||||
|
||||
Args:
|
||||
tasks (List[str]): A list of tasks to run.
|
||||
*args: Additional positional arguments for the task execution.
|
||||
**kwargs: Additional keyword arguments for the task execution.
|
||||
|
||||
Returns:
|
||||
List[Any]: The results of all executed tasks.
|
||||
|
||||
"""
|
||||
try:
|
||||
results = []
|
||||
with ThreadPoolExecutor(
|
||||
max_workers=self.max_workers
|
||||
) as executor:
|
||||
futures = [
|
||||
executor.submit(
|
||||
self.execute_task, task, *args, **kwargs
|
||||
)
|
||||
for task in tasks
|
||||
]
|
||||
for future in as_completed(futures):
|
||||
result = future.result()
|
||||
results.append(result)
|
||||
|
||||
return results
|
||||
except Exception as error:
|
||||
logger.error(f"Error in concurrent_run: {error}")
|
||||
return None
|
||||
@ -0,0 +1,326 @@
|
||||
import math
|
||||
import time
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from typing import Any, Callable, Dict, Optional, Tuple
|
||||
|
||||
from datasets import Dataset, load_dataset
|
||||
from loguru import logger
|
||||
from tqdm import tqdm
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Logging configuration: log to console and file (rotating by size)
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Swarm interface example
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Benchmark configuration
|
||||
# -----------------------------------------------------------------------------
|
||||
class BenchmarkConfig:
|
||||
"""
|
||||
Configuration for a benchmark dataset.
|
||||
|
||||
Attributes:
|
||||
input_column (str): The column containing the task prompt.
|
||||
answer_column (str): The column containing the expected answer.
|
||||
answer_extractor (Optional[Callable[[Any], str]]): Function to extract
|
||||
a string answer from the dataset's raw answer format.
|
||||
answer_matcher (Optional[Callable[[str, str], bool]]): Function to compare
|
||||
the expected answer and the swarm output. If None, a simple substring
|
||||
containment is used.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
input_column: str,
|
||||
answer_column: str,
|
||||
answer_extractor: Optional[Callable[[Any], str]] = None,
|
||||
answer_matcher: Optional[Callable[[str, str], bool]] = None,
|
||||
):
|
||||
self.input_column = input_column
|
||||
self.answer_column = answer_column
|
||||
self.answer_extractor = answer_extractor
|
||||
self.answer_matcher = answer_matcher
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Preset dataset configurations for popular benchmarks
|
||||
# -----------------------------------------------------------------------------
|
||||
PRESET_DATASETS: Dict[str, BenchmarkConfig] = {
|
||||
"gsm8k": BenchmarkConfig(
|
||||
input_column="question",
|
||||
answer_column="answer",
|
||||
),
|
||||
"squad": BenchmarkConfig(
|
||||
input_column="question",
|
||||
answer_column="answers",
|
||||
answer_extractor=lambda ans: (
|
||||
ans["text"][0]
|
||||
if isinstance(ans, dict)
|
||||
and "text" in ans
|
||||
and isinstance(ans["text"], list)
|
||||
and ans["text"]
|
||||
else str(ans)
|
||||
),
|
||||
),
|
||||
"winogrande": BenchmarkConfig(
|
||||
input_column="sentence",
|
||||
answer_column="answer",
|
||||
),
|
||||
"commonsense_qa": BenchmarkConfig(
|
||||
input_column="question",
|
||||
answer_column="answerKey",
|
||||
),
|
||||
# Add additional presets here.
|
||||
}
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# SwarmEvaluator with extended features
|
||||
# -----------------------------------------------------------------------------
|
||||
class SwarmEvaluator:
|
||||
"""
|
||||
Evaluator that uses a swarm of agents to process benchmark datasets
|
||||
from Hugging Face, with concurrency, retries, progress display, performance timing,
|
||||
and customizable answer matching.
|
||||
|
||||
Example:
|
||||
swarm = Swarm()
|
||||
evaluator = SwarmEvaluator(swarm)
|
||||
results = evaluator.evaluate("gsm8k", split="test", max_workers=4)
|
||||
print(results)
|
||||
"""
|
||||
|
||||
def __init__(self, swarm: callable) -> None:
|
||||
"""
|
||||
Initialize the evaluator with a given swarm.
|
||||
|
||||
Args:
|
||||
swarm (Swarm): A swarm instance with a callable run(task: str) method.
|
||||
"""
|
||||
self.swarm = swarm
|
||||
|
||||
def evaluate(
|
||||
self,
|
||||
dataset_name: str,
|
||||
split: str = "test",
|
||||
config: Optional[BenchmarkConfig] = None,
|
||||
max_workers: int = 1,
|
||||
max_retries: int = 3,
|
||||
show_progress: bool = True,
|
||||
output_file: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Evaluate the specified benchmark dataset using the swarm.
|
||||
|
||||
Args:
|
||||
dataset_name (str): The dataset name (from Hugging Face).
|
||||
split (str): The dataset split (e.g., "test", "validation").
|
||||
config (Optional[BenchmarkConfig]): Benchmark configuration. If None,
|
||||
a preset config is used.
|
||||
max_workers (int): Number of concurrent workers.
|
||||
max_retries (int): Number of retries for swarm tasks on failure.
|
||||
show_progress (bool): If True, display a progress bar.
|
||||
output_file (Optional[str]): Path to a file to write the results.
|
||||
|
||||
Returns:
|
||||
Dict[str, Any]: Evaluation metrics including total examples, correct answers,
|
||||
accuracy, and total evaluation time.
|
||||
"""
|
||||
if config is None:
|
||||
config = PRESET_DATASETS.get(dataset_name)
|
||||
if config is None:
|
||||
raise ValueError(
|
||||
f"No preset config for dataset '{dataset_name}'. Provide a BenchmarkConfig."
|
||||
)
|
||||
|
||||
logger.info(
|
||||
f"Loading dataset '{dataset_name}' (split: {split})..."
|
||||
)
|
||||
dataset: Dataset = load_dataset(dataset_name, split=split)
|
||||
total_examples = len(dataset)
|
||||
logger.info(f"Total examples to evaluate: {total_examples}")
|
||||
|
||||
start_time = time.time()
|
||||
correct = 0
|
||||
|
||||
# Function to process a single example.
|
||||
def _process_example(
|
||||
example: Dict[str, Any], idx: int
|
||||
) -> Tuple[bool, float]:
|
||||
task_start = time.time()
|
||||
task_text = example.get(config.input_column)
|
||||
expected_answer = example.get(config.answer_column)
|
||||
|
||||
if task_text is None or expected_answer is None:
|
||||
logger.warning(
|
||||
f"Example {idx}: Missing '{config.input_column}' or '{config.answer_column}', skipping."
|
||||
)
|
||||
return (False, 0.0)
|
||||
|
||||
# Use answer_extractor if provided.
|
||||
if config.answer_extractor:
|
||||
try:
|
||||
expected_answer = config.answer_extractor(
|
||||
expected_answer
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Example {idx}: Error extracting answer: {e}"
|
||||
)
|
||||
return (False, 0.0)
|
||||
|
||||
logger.debug(f"Example {idx} - Task: {task_text}")
|
||||
logger.debug(
|
||||
f"Example {idx} - Expected Answer: {expected_answer}"
|
||||
)
|
||||
|
||||
try:
|
||||
swarm_output = self._run_with_retry(
|
||||
task_text, max_retries
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Example {idx}: Failed after retries. Error: {e}"
|
||||
)
|
||||
return (False, time.time() - task_start)
|
||||
|
||||
logger.debug(
|
||||
f"Example {idx} - Swarm Output: {swarm_output}"
|
||||
)
|
||||
|
||||
# Use custom matcher if provided; otherwise, default matching.
|
||||
if config.answer_matcher:
|
||||
is_correct = config.answer_matcher(
|
||||
expected_answer, swarm_output
|
||||
)
|
||||
else:
|
||||
is_correct = self._default_matcher(
|
||||
expected_answer, swarm_output
|
||||
)
|
||||
|
||||
task_time = time.time() - task_start
|
||||
logger.info(
|
||||
f"Example {idx}: {'Correct' if is_correct else 'Incorrect'} in {task_time:.2f}s"
|
||||
)
|
||||
return (is_correct, task_time)
|
||||
|
||||
# Use ThreadPoolExecutor for concurrency.
|
||||
futures = []
|
||||
total_time = 0.0
|
||||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||||
# Optionally wrap the dataset with tqdm for a progress bar.
|
||||
examples_iter = enumerate(dataset, start=1)
|
||||
if show_progress:
|
||||
examples_iter = tqdm(
|
||||
list(examples_iter),
|
||||
total=total_examples,
|
||||
desc="Evaluating",
|
||||
)
|
||||
|
||||
for idx, example in examples_iter:
|
||||
futures.append(
|
||||
executor.submit(_process_example, example, idx)
|
||||
)
|
||||
|
||||
for future in as_completed(futures):
|
||||
try:
|
||||
is_correct, elapsed = future.result()
|
||||
total_time += elapsed
|
||||
if is_correct:
|
||||
correct += 1
|
||||
except Exception as e:
|
||||
logger.error(f"Error processing an example: {e}")
|
||||
|
||||
overall_time = time.time() - start_time
|
||||
accuracy = (
|
||||
correct / total_examples if total_examples > 0 else 0.0
|
||||
)
|
||||
|
||||
logger.info(
|
||||
f"Evaluation complete. Total examples: {total_examples}, Correct: {correct}, "
|
||||
f"Accuracy: {accuracy:.2%}, Overall Time: {overall_time:.2f}s, "
|
||||
f"Average per-example time: {total_time/total_examples if total_examples else 0:.2f}s"
|
||||
)
|
||||
|
||||
results = {
|
||||
"total": total_examples,
|
||||
"correct": correct,
|
||||
"accuracy": accuracy,
|
||||
"overall_time": overall_time,
|
||||
"average_example_time": (
|
||||
total_time / total_examples
|
||||
if total_examples
|
||||
else math.nan
|
||||
),
|
||||
}
|
||||
|
||||
# Optionally save results to a file.
|
||||
if output_file:
|
||||
try:
|
||||
with open(output_file, "w") as f:
|
||||
for key, value in results.items():
|
||||
f.write(f"{key}: {value}\n")
|
||||
logger.info(f"Results saved to {output_file}")
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error saving results to {output_file}: {e}"
|
||||
)
|
||||
|
||||
return results
|
||||
|
||||
def _run_with_retry(self, task: str, max_retries: int) -> str:
|
||||
"""
|
||||
Runs the swarm task with a retry mechanism.
|
||||
|
||||
Args:
|
||||
task (str): The task string.
|
||||
max_retries (int): Maximum number of retries.
|
||||
|
||||
Returns:
|
||||
str: Swarm output.
|
||||
|
||||
Raises:
|
||||
Exception: If all retries fail.
|
||||
"""
|
||||
attempt = 0
|
||||
while attempt <= max_retries:
|
||||
try:
|
||||
start = time.time()
|
||||
result = self.swarm.run(task)
|
||||
elapsed = time.time() - start
|
||||
logger.debug(
|
||||
f"Task succeeded in {elapsed:.2f}s on attempt {attempt + 1}"
|
||||
)
|
||||
return result
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Task failed on attempt {attempt + 1}: {e}"
|
||||
)
|
||||
attempt += 1
|
||||
time.sleep(0.5 * attempt) # Exponential backoff
|
||||
raise Exception("Max retries exceeded for task.")
|
||||
|
||||
@staticmethod
|
||||
def _default_matcher(expected: str, output: str) -> bool:
|
||||
"""
|
||||
Default answer matching using a normalized substring check.
|
||||
|
||||
Args:
|
||||
expected (str): The expected answer.
|
||||
output (str): The swarm output.
|
||||
|
||||
Returns:
|
||||
bool: True if expected is found in output; otherwise, False.
|
||||
"""
|
||||
expected_norm = " ".join(expected.strip().split())
|
||||
output_norm = " ".join(output.strip().split())
|
||||
return expected_norm in output_norm
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Example usage
|
||||
# -----------------------------------------------------------------------------
|
||||
@ -1,177 +0,0 @@
|
||||
import asyncio
|
||||
import time
|
||||
from swarms.structs.agent import Agent
|
||||
from swarms.structs.multi_process_workflow import MultiProcessWorkflow
|
||||
|
||||
|
||||
def create_test_agent(name: str) -> Agent:
|
||||
"""Create a test agent that simply returns its input with a timestamp"""
|
||||
return Agent(
|
||||
agent_name=name,
|
||||
system_prompt=f"Test prompt for {name}",
|
||||
model_name="gpt-4o-mini",
|
||||
max_loops=1,
|
||||
)
|
||||
|
||||
|
||||
def test_initialization():
|
||||
"""Test basic workflow initialization"""
|
||||
print("\n=== Testing Workflow Initialization ===")
|
||||
try:
|
||||
agents = [create_test_agent(f"agent{i}") for i in range(3)]
|
||||
workflow = MultiProcessWorkflow(max_workers=2, agents=agents)
|
||||
|
||||
print("✓ Created workflow with configuration:")
|
||||
print(f" - Max workers: {workflow.max_workers}")
|
||||
print(f" - Number of agents: {len(workflow.agents)}")
|
||||
print(f" - Autosave: {workflow.autosave}")
|
||||
print("✓ Initialization test passed")
|
||||
except Exception as e:
|
||||
print(f"✗ Initialization test failed: {str(e)}")
|
||||
raise
|
||||
|
||||
|
||||
def test_execute_task():
|
||||
"""Test execution of a single task"""
|
||||
print("\n=== Testing Task Execution ===")
|
||||
try:
|
||||
agents = [create_test_agent("test_agent")]
|
||||
workflow = MultiProcessWorkflow(agents=agents)
|
||||
|
||||
test_task = "Return this message with timestamp"
|
||||
result = workflow.execute_task(test_task)
|
||||
|
||||
print("✓ Task executed successfully")
|
||||
print(f" - Input task: {test_task}")
|
||||
print(f" - Result: {result}")
|
||||
print("✓ Task execution test passed")
|
||||
except Exception as e:
|
||||
print(f"✗ Task execution test failed: {str(e)}")
|
||||
raise
|
||||
|
||||
|
||||
def test_parallel_run():
|
||||
"""Test parallel execution of tasks"""
|
||||
print("\n=== Testing Parallel Run ===")
|
||||
try:
|
||||
agents = [create_test_agent(f"agent{i}") for i in range(3)]
|
||||
workflow = MultiProcessWorkflow(max_workers=2, agents=agents)
|
||||
|
||||
test_task = "Process this in parallel"
|
||||
results = workflow.run(test_task)
|
||||
|
||||
print("✓ Parallel execution completed")
|
||||
# print(f" - Number of results: {len(results)}")
|
||||
print(f" - Results: {results}")
|
||||
print("✓ Parallel run test passed")
|
||||
except Exception as e:
|
||||
print(f"✗ Parallel run test failed: {str(e)}")
|
||||
raise
|
||||
|
||||
|
||||
async def test_async_run():
|
||||
"""Test asynchronous execution of tasks"""
|
||||
print("\n=== Testing Async Run ===")
|
||||
try:
|
||||
agents = [create_test_agent(f"agent{i}") for i in range(3)]
|
||||
workflow = MultiProcessWorkflow(max_workers=2, agents=agents)
|
||||
|
||||
test_task = "Process this asynchronously"
|
||||
results = await workflow.async_run(test_task)
|
||||
|
||||
print("✓ Async execution completed")
|
||||
print(f" - Number of results: {len(results)}")
|
||||
print(f" - Results: {results}")
|
||||
print("✓ Async run test passed")
|
||||
except Exception as e:
|
||||
print(f"✗ Async run test failed: {str(e)}")
|
||||
raise
|
||||
|
||||
|
||||
def test_batched_run():
|
||||
"""Test batch execution of tasks"""
|
||||
print("\n=== Testing Batched Run ===")
|
||||
try:
|
||||
agents = [create_test_agent(f"agent{i}") for i in range(2)]
|
||||
workflow = MultiProcessWorkflow(max_workers=2, agents=agents)
|
||||
|
||||
tasks = [f"Batch task {i}" for i in range(5)]
|
||||
results = workflow.batched_run(tasks, batch_size=2)
|
||||
|
||||
print("✓ Batch execution completed")
|
||||
print(f" - Number of tasks: {len(tasks)}")
|
||||
print(" - Batch size: 2")
|
||||
print(f" - Results: {results}")
|
||||
print("✓ Batched run test passed")
|
||||
except Exception as e:
|
||||
print(f"✗ Batched run test failed: {str(e)}")
|
||||
raise
|
||||
|
||||
|
||||
def test_concurrent_run():
|
||||
"""Test concurrent execution of tasks"""
|
||||
print("\n=== Testing Concurrent Run ===")
|
||||
try:
|
||||
agents = [create_test_agent(f"agent{i}") for i in range(2)]
|
||||
workflow = MultiProcessWorkflow(max_workers=2, agents=agents)
|
||||
|
||||
tasks = [f"Concurrent task {i}" for i in range(4)]
|
||||
results = workflow.concurrent_run(tasks)
|
||||
|
||||
print("✓ Concurrent execution completed")
|
||||
print(f" - Number of tasks: {len(tasks)}")
|
||||
print(f" - Results: {results}")
|
||||
print("✓ Concurrent run test passed")
|
||||
except Exception as e:
|
||||
print(f"✗ Concurrent run test failed: {str(e)}")
|
||||
raise
|
||||
|
||||
|
||||
def test_error_handling():
|
||||
"""Test error handling in workflow"""
|
||||
print("\n=== Testing Error Handling ===")
|
||||
try:
|
||||
# Create workflow with no agents to trigger error
|
||||
workflow = MultiProcessWorkflow(max_workers=2, agents=None)
|
||||
result = workflow.execute_task(
|
||||
"This should handle the error gracefully"
|
||||
)
|
||||
|
||||
print("✓ Error handled gracefully")
|
||||
print(f" - Result when no agents: {result}")
|
||||
print("✓ Error handling test passed")
|
||||
except Exception as e:
|
||||
print(f"✗ Error handling test failed: {str(e)}")
|
||||
raise
|
||||
|
||||
|
||||
async def run_all_tests():
|
||||
"""Run all tests"""
|
||||
print("\n=== Starting MultiProcessWorkflow Test Suite ===")
|
||||
start_time = time.time()
|
||||
|
||||
try:
|
||||
# Run synchronous tests
|
||||
test_initialization()
|
||||
test_execute_task()
|
||||
test_parallel_run()
|
||||
test_batched_run()
|
||||
test_concurrent_run()
|
||||
test_error_handling()
|
||||
|
||||
# Run async test
|
||||
await test_async_run()
|
||||
|
||||
end_time = time.time()
|
||||
duration = round(end_time - start_time, 2)
|
||||
print("\n=== Test Suite Completed Successfully ===")
|
||||
print(f"Time taken: {duration} seconds")
|
||||
|
||||
except Exception as e:
|
||||
print("\n=== Test Suite Failed ===")
|
||||
print(f"Error: {str(e)}")
|
||||
raise
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(run_all_tests())
|
||||
Loading…
Reference in new issue