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

spreadsheet and syntax fix to auto_test_eval.py

Browse Source
pull/700/head
Patrick Devaney 1 week ago
parent c9f69d5fef
commit 02a6fbc694
6 changed files with 160 additions and 192 deletions
Show Stats Download Patch File Download Diff File

82
Dockerfile
Unescape View File

@ -1,31 +1,55 @@
# Use an official Python runtime as a parent image
FROM python:3.11-slim
# Use Python 3.11 slim-bullseye for smaller base image
FROM python:3.11-slim-bullseye AS builder
# Set environment variables
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1
# Set the working directory in the container
WORKDIR /usr/src/swarms
# Install Python dependencies
# COPY requirements.txt and pyproject.toml if you're using poetry for dependency management
COPY requirements.txt .
RUN pip install --upgrade pip
RUN pip install --no-cache-dir -r requirements.txt
# Install the 'swarms' package, assuming it's available on PyPI
RUN pip install -U swarms
# Copy the rest of the application
COPY . .
# Expose port if your application has a web interface
# EXPOSE 5000
# # Define environment variable for the swarm to work
# ENV OPENAI_API_KEY=your_swarm_api_key_here
# If you're using `CMD` to execute a Python script, make sure it's executable
# RUN chmod +x example.py
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PIP_NO_CACHE_DIR=1 \
PIP_DISABLE_PIP_VERSION_CHECK=1
# Set the working directory
WORKDIR /build
# Install only essential build dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
build-essential \
gcc \
g++ \
gfortran \
&& rm -rf /var/lib/apt/lists/*
# Install swarms packages
RUN pip install --no-cache-dir swarm-models swarms
# Production stage
FROM python:3.11-slim-bullseye
# Set secure environment variables
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
WORKSPACE_DIR="agent_workspace" \
PATH="/app:${PATH}" \
PYTHONPATH="/app:${PYTHONPATH}" \
USER=swarms
# Create non-root user
RUN useradd -m -s /bin/bash -U $USER && \
mkdir -p /app && \
chown -R $USER:$USER /app
# Set working directory
WORKDIR /app
# Copy only necessary files from builder
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY --from=builder /usr/local/bin /usr/local/bin
# Copy application with correct permissions
COPY --chown=$USER:$USER . .
# Switch to non-root user
USER $USER
# Health check
HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
CMD python -c "import swarms; print('Health check passed')" || exit 1

2
pyproject.toml
Unescape View File

@ -80,7 +80,7 @@ rich = "*"
# sentence-transformers = "*"
swarm-models = "*"
termcolor = "*"
clusterops = { git = "https://github.com/patrickbdevaney/clusterops.git", branch = "main" }
clusterops = { git = "https://github.com/The-Swarm-Corporation/clusterops.git", branch = "main" }
# [tool.poetry.extras]

2
requirements.txt
Unescape View File

@ -24,7 +24,7 @@ pytest>=8.1.1
pandas>=2.2.2
networkx
aiofiles
clusterops
git+https://github.com/The-Swarm-Corporation/clusterops.git@main#egg=clusterops
reportlab
doc-master
termcolor

56
tests/Dockerfile
Unescape View File

@ -20,7 +20,61 @@ COPY . .
RUN pip install poetry
# Disable virtualenv creation by poetry and install dependencies
RUN poetry config virtualenvs.create false
RUN poetry config vir# Use Python 3.11 slim-bullseye for smaller base image
FROM python:3.11-slim-bullseye AS builder
# Set environment variables
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PIP_NO_CACHE_DIR=1 \
PIP_DISABLE_PIP_VERSION_CHECK=1
# Set the working directory
WORKDIR /build
# Install only essential build dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
build-essential \
gcc \
g++ \
gfortran \
&& rm -rf /var/lib/apt/lists/*
# Install swarms packages
RUN pip install --no-cache-dir swarm-models swarms
# Production stage
FROM python:3.11-slim-bullseye
# Set secure environment variables
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
WORKSPACE_DIR="agent_workspace" \
PATH="/app:${PATH}" \
PYTHONPATH="/app:${PYTHONPATH}" \
USER=swarms
# Create non-root user
RUN useradd -m -s /bin/bash -U $USER && \
mkdir -p /app && \
chown -R $USER:$USER /app
# Set working directory
WORKDIR /app
# Copy only necessary files from builder
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY --from=builder /usr/local/bin /usr/local/bin
# Copy application with correct permissions
COPY --chown=$USER:$USER . .
# Switch to non-root user
USER $USER
# Health check
HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
CMD python -c "import swarms; print('Health check passed')" || exit 1tualenvs.create false
# Install the 'swarms' package if it's not included in the poetry.lock
RUN pip install swarms

145
tests/agent_evals/auto_test_eval.py
Unescape View File

@ -14,7 +14,6 @@ from swarm_models import OpenAIChat
from swarms.structs.agent import Agent
@dataclass
class SwarmSystemInfo:
"""System information for Swarms issue reports."""
@ -93,45 +92,33 @@ class SwarmsIssueReporter:
self.last_issue_time = datetime.now()
def _get_swarms_version(self) -> str:
"""Get the installed version of Swarms."""
try:
import swarms
"""Get the installed version of Swarms."""
try:
import swarms
return swarms.__version__
except:
return "Unknown"
return swarms.__version__
except:
return "Unknown"
def _get_gpu_info(self) -> Tuple[bool, Optional[str]]:
"""Get GPU information and CUDA availability."""
try:
import torch
cuda_available = torch.cuda.is_available()
if cuda_available:
gpu_info = torch.cuda.get_device_name(0)
return cuda_available, gpu_info
return False, None
except ModuleNotFoundError as e:
# Handle the case where 'torch' is not installed
print(f"Error: {e}")
return False, None
except RuntimeError as e:
# Handle CUDA-related errors
print(f"Error: {e}")def _get_swarms_version(self) -> str:
"""Get the installed version of Swarms."""
try:
import swarms
return swarms.__version__
except:
return "Unknown"
return False, None
except Exception as e:
# Catch any other exceptions
print(f"Unexpected error: {e}")
return False, None
"""Get GPU information and CUDA availability."""
try:
import torch
cuda_available = torch.cuda.is_available()
if cuda_available:
gpu_info = torch.cuda.get_device_name(0)
return cuda_available, gpu_info
return False, None
except ModuleNotFoundError as e:
print(f"Error: {e}")
return False, None
except RuntimeError as e:
print(f"Error: {e}")
return False, None
except Exception as e:
print(f"Unexpected error: {e}")
return False, None
def _get_system_info(self) -> SwarmSystemInfo:
"""Collect system and Swarms-specific information."""
@ -149,23 +136,15 @@ class SwarmsIssueReporter:
gpu_info=gpu_info,
)
def _categorize_error(
self, error: Exception, context: Dict
) -> List[str]:
def _categorize_error(self, error: Exception, context: Dict) -> List[str]:
"""Categorize the error and return appropriate labels."""
error_str = str(error).lower()
type(error).__name__
labels = ["bug", "automated"]
# Check error message and context for category keywords
for (
category,
category_labels,
) in self.ISSUE_CATEGORIES.items():
if any(
keyword in error_str for keyword in category_labels
):
for category, category_labels in self.ISSUE_CATEGORIES.items():
if any(keyword in error_str for keyword in category_labels):
labels.extend(category_labels)
break
@ -180,10 +159,7 @@ class SwarmsIssueReporter:
return list(set(labels)) # Remove duplicates
def _format_swarms_issue_body(
self,
error: Exception,
system_info: SwarmSystemInfo,
context: Dict,
self, error: Exception, system_info: SwarmSystemInfo, context: Dict
) -> str:
"""Format the issue body with Swarms-specific information."""
return f"""
@ -218,42 +194,33 @@ class SwarmsIssueReporter:
"""
def _get_dependencies_info(self) -> str:
"""Get information about installed dependencies."""
try:
import pkg_resources
deps = []
for dist in pkg_resources.working_set:
deps.append(f"- {dist.key} {dist.version}")
return "\n".join(deps)
except ImportError as e:
# Handle the case where pkg_resources is not available
print(f"Error: {e}")
return "Unable to fetch dependency information"
except Exception as e:
# Catch any other unexpected exceptions
print(f"Unexpected error: {e}")
return "Unable to fetch dependency information"
# First, add this method to your SwarmsIssueReporter class
"""Get information about installed dependencies."""
try:
import pkg_resources
deps = []
for dist in pkg_resources.working_set:
deps.append(f"- {dist.key} {dist.version}")
return "\n".join(deps)
except ImportError as e:
print(f"Error: {e}")
return "Unable to fetch dependency information"
except Exception as e:
print(f"Unexpected error: {e}")
return "Unable to fetch dependency information"
def _check_rate_limit(self) -> bool:
"""Check if we're within rate limits."""
now = datetime.now()
time_diff = (now - self.last_issue_time).total_seconds()
if (
len(self.issues_created) >= self.rate_limit
and time_diff < self.rate_period
):
if len(self.issues_created) >= self.rate_limit and time_diff < self.rate_period:
logger.warning("Rate limit exceeded for issue creation")
return False
# Clean up old issues from tracking
self.issues_created = [
time
for time in self.issues_created
if (now - time).total_seconds() < self.rate_period
time for time in self.issues_created if (now - time).total_seconds() < self.rate_period
]
return True
@ -279,9 +246,7 @@ class SwarmsIssueReporter:
"""
try:
if not self._check_rate_limit():
logger.warning(
"Skipping issue creation due to rate limit"
)
logger.warning("Skipping issue creation due to rate limit")
return None
# Collect system information
@ -312,25 +277,19 @@ class SwarmsIssueReporter:
url = f"https://api.github.com/repos/{self.REPO_OWNER}/{self.REPO_NAME}/issues"
data = {
"title": title,
"body": self._format_swarms_issue_body(
error, system_info, full_context
),
"body": self._format_swarms_issue_body(error, system_info, full_context),
"labels": labels,
}
response = requests.post(
url,
headers={
"Authorization": f"token {self.github_token}"
},
headers={"Authorization": f"token {self.github_token}"},
json=data,
)
response.raise_for_status()
issue_number = response.json()["number"]
logger.info(
f"Successfully created Swarms issue #{issue_number}"
)
logger.info(f"Successfully created Swarms issue #{issue_number}")
return issue_number
@ -340,15 +299,11 @@ class SwarmsIssueReporter:
# Setup the reporter with your GitHub token
reporter = SwarmsIssueReporter(
github_token=os.getenv("GITHUB_API_KEY")
)
reporter = SwarmsIssueReporter(github_token=os.getenv("GITHUB_API_KEY"))
# Force an error to test the reporter
try:
# This will raise an error since the input isn't valid
# Create an agent that might have issues
model = OpenAIChat(model_name="gpt-4o")
agent = Agent(agent_name="Test-Agent", max_loops=1)

65
tests/structs/test_spreadsheet_swarm_2.py
Unescape View File

@ -1,65 +0,0 @@
import os
from datetime import datetime
from uuid import uuid4
# Import necessary classes from your swarm module
from swarms.structs.agent import Agent
from swarms.structs.base_swarm import BaseSwarm
from swarms.telemetry.capture_sys_data import log_agent_data
from swarms.utils.file_processing import create_file_in_folder
from swarms import SpreadSheetSwarm
# Ensure you have an environment variable or default workspace dir
workspace_dir = os.getenv("WORKSPACE_DIR", "./workspace")
def create_agents(num_agents: int):
"""
Create a list of agent instances.
Args:
num_agents (int): The number of agents to create.
Returns:
List[Agent]: List of created Agent objects.
"""
agents = []
for i in range(num_agents):
agent_name = f"Agent-{i + 1}"
agents.append(Agent(agent_name=agent_name))
return agents
def main():
# Number of agents to create
num_agents = 5
# Create the agents
agents = create_agents(num_agents)
# Initialize the swarm with agents and other configurations
swarm = SpreadSheetSwarm(
name="Test-Swarm",
description="A swarm for testing purposes.",
agents=agents,
autosave_on=True,
max_loops=2,
workspace_dir=workspace_dir
)
# Run a sample task in the swarm (synchronously)
task = "process_data"
# Ensure the run method is synchronous
swarm_metadata = swarm.run(task) # Assuming this is made synchronous
# Print swarm metadata after task completion
print("Swarm Metadata:")
print(swarm_metadata)
# Check if CSV file has been created and saved
if os.path.exists(swarm.save_file_path):
print(f"Metadata saved to: {swarm.save_file_path}")
else:
print(f"Metadata not saved correctly. Check the save path.")
# Test saving metadata to JSON file
swarm.data_to_json_file()
# Test exporting metadata to JSON
swarm_json = swarm.export_to_json()
print("Exported JSON metadata:")
print(swarm_json)
# Log agent data
print("Logging agent data:")
print(log_agent_data(swarm.metadata.model_dump()))
# Run the synchronous main function
if __name__ == "__main__":
main()
Write Preview
Loading…
Cancel
Save