abstractswarm

swarms/agents/__init__.py

@@ -11,3 +11,4 @@ from swarms.agents.omni_modal_agent import OmniModalAgent
 from swarms.agents.message import Message
 from swarms.agents.stream_response import stream
 from swarms.agents.base import AbstractAgent
+from swarms.agents.registry import Registry

swarms/agents/registry.py

@@ -0,0 +1,27 @@
+from typing import Dict
+
+from pydantic import BaseModel
+
+
+class Registry(BaseModel):
+    """Registry for storing and building classes."""
+
+    name: str
+    entries: Dict = {}
+
+    def register(self, key: str):
+        def decorator(class_builder):
+            self.entries[key] = class_builder
+            return class_builder
+
+        return decorator
+
+    def build(self, type: str, **kwargs):
+        if type not in self.entries:
+            raise ValueError(
+                f'{type} is not registered. Please register with the .register("{type}") method provided in {self.name} registry'
+            )
+        return self.entries[type](**kwargs)
+
+    def get_all_entries(self):
+        return self.entries

swarms/prompts/python.py

@@ -232,4 +232,3 @@ PY_TEST_GENERATION_COMPLETION_INSTRUCTION = f"""You are an AI coding assistant t
 {PY_TEST_GENERATION_FEW_SHOT}"""
 
 PY_TEST_GENERATION_CHAT_INSTRUCTION = """You are an AI coding assistant that can write unique, diverse, and intuitive unit tests for functions given the signature and docstring."""
-

swarms/structs/nonlinear_workflow.py

@@ -7,6 +7,7 @@ class Task:
     """
     Task is a unit of work that can be executed by an agent
     """
+
     def __init__(
         self, id: str, parents: List["Task"] = None, children: List["Task"] = None
     ):

swarms/swarms/base.py

@@ -1,5 +1,6 @@
 from abc import ABC, abstractmethod
-
+from typing import Optional, List
+from swarms.workers.base import AbstractWorker
 
 class AbstractSwarm(ABC):
     # TODO: Pass in abstract LLM class that can utilize Hf or Anthropic models, Move away from OPENAI
@@ -8,15 +9,52 @@ class AbstractSwarm(ABC):
     # TODO: Add RLHF Data collection, ask user how the swarm is performing
     # TODO: Create an onboarding process if not settings are preconfigured like `from swarms import Swarm, Swarm()` => then initiate onboarding name your swarm + provide purpose + etc
 
-    def __init__(self, agents, vectorstore, tools):
+    @abstractmethod
-        self.agents = agents
+    def __init__(self, workers: List["AbstractWorker"]):
-        self.vectorstore = vectorstore
+        """Initialize the swarm with workers"""
-        self.tools = tools
+        pass
 
     @abstractmethod
     def communicate(self):
+        """Communicate with the swarm through the orchestrator, protocols, and the universal communication layer"""
         pass
 
     @abstractmethod
     def run(self):
+        """Run the swarm"""
+        pass
+
+    @abstractmethod
+    def add_worker(self, worker: "AbstractWorker"):
+        """Add a worker to the swarm"""
+        pass
+
+    @abstractmethod
+    def remove_worker(self, worker: "AbstractWorker"):
+        """Remove a worker from the swarm"""
+        pass
+
+    @abstractmethod
+    def broadcast(self, message: str, sender: Optional["AbstractWorker"] = None):
+        """Broadcast a message to all workers"""
+        pass
+
+    @abstractmethod
+    def reset(self):
+        """Reset the swarm"""
+        pass
+
+    @abstractmethod
+    def plan(self, task: str):
+        """Workers must individually plan using a workflow or pipeline"""
+        pass
+
+    @abstractmethod
+    def direct_message(
+        self,
+        message: str,
+        sender: "AbstractWorker",
+        recipient: "AbstractWorker",
+    ):
+        """Send a direct message to a worker"""
         pass