`Task`

Former-commit-id: 88c6d6343f

pyproject.toml

@@ -46,6 +46,7 @@ redis = "*"
 Pillow = "*"
 shapeless="*"
 chromadb = "*"
+agent-protocol = "*"
 
 [tool.poetry.dev-dependencies]
 # Add development dependencies here

requirements.txt

@@ -18,6 +18,9 @@ openai
 google-generativeai
 duckduckgo-search
 shapeless
+agent-protocol
+
+
 
 mkdocs
 mkdocs-material

swarms/agents/base.py

@@ -1,4 +1,6 @@
 from abc import ABC, abstractmethod
+from agent_protocol import Agent, Step, Task
+
 
 class AbstractAgent(ABC):
     #absrtact agent class

swarms/artifacts/error_artifact.py

@@ -1,16 +0,0 @@
-# #from shapeless import shapeless
-
-# #@shapeless
-# class ErrorArtifact:
-#     def __init__(
-#         self,
-#         value
-#     ):
-#         self.value = value
-    
-#     def __add__(self, other):
-#         return ErrorArtififact(self.value + other.value)
-    
-#     def to_text(self) -> str:
-#         return self.value
-    

swarms/artifacts/main.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+import pprint
+import json
+
+from typing import Optional
+from pydantic import BaseModel, Field, StrictStr
+
+class Artifact(BaseModel):
+    """
+
+    Artifact that has the task has been produced
+    """
+
+    artifact_id: StrictStr = Field(
+        ...,
+        description="ID of the artifact"
+    )
+    file_name: StrictStr = Field(
+        ...,
+        description="Filename of the artifact"
+    )
+    relative_path: Optional[StrictStr] = Field(
+        None, description="Relative path of the artifact"
+    )
+    __properties = ["artifact_id", "file_name", "relative_path"]
+
+    class Config:
+        """Pydantic configuration"""
+
+        allow_population_by_field_name = True
+        validate_assignment = True
+
+    def to_str(self) -> str:
+        """Returns the string representation of the model using alias"""
+        return pprint.pformat(self.dict(by_alias=True))
+    
+    @classmethod
+    def from_json(cls, json_str: str) -> Artifact:
+        """Create an instance of Artifact from a json string"""
+        return cls.from_dict(json.loads(json_str))
+    
+    def to_dict(self):
+        """Returns the dict representation of the model"""
+        _dict = self.dict(by_alias=True, exclude={}, exclude_none=True)
+        return _dict
+    
+    @classmethod
+    def from_dict(cls, obj: dict) -> Artifact:
+        """Create an instance of Artifact from a dict"""
+        
+        if obj is None:
+            return None
+        
+        if not isinstance(obj, dict):
+            return Artifact.parse_obj(obj)
+        
+        _obj = Artifact.parse_obj(
+            {
+                "artifact_id": obj.get("artifact_id"),
+                "file_name": obj.get("file_name"),
+                "relative_path": obj.get("relative_path"),
+            }
+        )
+
+        return _obj
+
+

swarms/structs/task.py

@@ -1,44 +1,84 @@
-from abc import ABC
+from __future__ import annotations
-#from shapeless import shapeless
+import pprint
+import json
 
+from typing import Optional, Any
+from pydantic import BaseModel, Field, StrictStr, StrictStr, conlist
+from artifacts.main import Artifact
 
-#@shapeless
+class Task(BaseModel):
-class BaseTask(ABC):
+    input: Optional[StrictStr] = Field(
-    def __init__(
+        None,
-        self,
+        description="Input prompt for the task"
-        id, 
+    )
-    ):
+    additional_input: Optional[Any] = Field(
-        self.id = id
+        None,
+        description="Input parameters for the task. Any value is allowed"
+    )
+    task_id: StrictStr = Field(
+        ...,
+        description="ID of the task"
+    )
+    artifacts: conlist(Artifact) = Field(
+        ...,
+        description="A list of artifacts that the task has been produced"
+    )
 
-    def add(self):
+    __properties = ["input", "additional_input", "task_id", "artifact"]
-        pass
-    
-    def schedule(self, time):
-        pass
-
-    def parents(self):
-        pass
-
-    def children(self):
-        pass
 
-    def preprocess(self):
+    class Config:
-        pass
+        #pydantic config
+        
+        allow_population_by_field_name = True
+        validate_assignment = True
 
-    def add_parent(self):
+    def to_str(self) -> str:
-        pass
+        """Returns the str representation of the model using alias"""
+        return pprint.pformat(self.dict(by_alias=True))
     
-    def is_pending(self):
+    def to_json(self) -> str:
-        pass
+        """Returns the JSON representation of the model using alias"""
-
+        return json.dumps(self.to_dict())
-    def is_finished(self):
+    
-        pass
+    @classmethod
-
+    def from_json(cls, json_str: str) -> Task:
-    def is_executing(self):
+        """Create an instance of Task from a json string"""
-        pass
+        return cls.from_dict(json.loads(json_str))
-
+    
-    def run(self):
+    def to_dict(self):
-        pass
+        """Returns the dict representation of the model using alias"""
-
+        _dict = self.dict(by_alias=True, exclude={}, exclude_none=True)
-    def reset(self):
+        _items =[]
-        pass
+        if self.artifacts:
+            for _item in self.artifacts:
+                if _item:
+                    _items.append(_item.to_dict())
+            _dict["artifacts"] = _items
+        #set to None if additional input is None
+        # and __fields__set contains the field
+        if self.additional_input is None and "additional_input" in self.__fields__set__:
+            _dict["additional_input"] = None
+        
+        return _dict
+    
+    @classmethod
+    def from_dict(cls, obj: dict) -> Task:
+        """Create an instance of Task from dict"""
+        if obj is None:
+            return None
+        
+        if not isinstance(obj, dict):
+            return Task.parse_obj(obj)
+        
+        _obj = Task.parse_obj(
+            {
+                "input": obj.get("input"),
+                "additional_input": obj.get("additional_input"),
+                "task_id": obj.get("task_id"),
+                "artifacts": [
+                    Artifact.from_dict(_item) for _item in obj.get("artifacts")
+                ]
+                if obj.get("artifacts") is not None
+                else None,
+            }
+        )