CLEAN UP: Flow and demo layouts

swarms/structs/flow.py

@@ -12,13 +12,7 @@
 from swarms.utils.code_interpreter import SubprocessCodeInterpreter
 from swarms.utils.parse_code import extract_code_in_backticks_in_string
 
-# Prompts
-DYNAMIC_STOP_PROMPT = """
-When you have finished the task from the Human, output a special token: <DONE>
-This will enable you to leave the autonomous loop.
-"""
-
-# Constants
+# System prompt
 FLOW_SYSTEM_PROMPT = f"""
 You are an autonomous agent granted autonomy in a autonomous loop structure.
 Your role is to engage in multi-step conversations with your self or the user,
@@ -30,6 +24,19 @@
 
 """
 
+
+
+# Prompts
+DYNAMIC_STOP_PROMPT = """
+
+Now, when you 99% sure you have completed the task, you may follow the instructions below to escape the autonomous loop.
+
+When you have finished the task from the Human, output a special token: <DONE>
+This will enable you to leave the autonomous loop.
+"""
+
+
+
 # Make it able to handle multi input tools
 DYNAMICAL_TOOL_USAGE = """
 You have access to the following tools:
@@ -191,7 +198,7 @@ class Flow:
     def __init__(
         self,
         llm: Any,
-        # template: str,
+        template: str,
         max_loops=5,
         stopping_condition: Optional[Callable[[str], bool]] = None,
         loop_interval: int = 1,
@@ -217,6 +224,7 @@ def __init__(
         **kwargs: Any,
     ):
         self.llm = llm
+        self.template = template
         self.max_loops = max_loops
         self.stopping_condition = stopping_condition
         self.loop_interval = loop_interval

swarms/structs/non_linear_workflow.py

@@ -0,0 +1,79 @@
+from swarms.models import OpenAIChat
+from swarms.structs.flow import Flow
+
+import concurrent.futures
+from typing import Callable, List, Dict, Any, Sequence
+
+
+class Task:
+    def __init__(self, id: str, task: str, flows: Sequence[Flow], dependencies: List[str] = []):
+        self.id = id
+        self.task = task
+        self.flows = flows
+        self.dependencies = dependencies
+        self.results = []
+
+    def execute(self, parent_results: Dict[str, Any]):
+        args = [parent_results[dep] for dep in self.dependencies]
+        for flow in self.flows:
+            result = flow.run(self.task, *args)
+            self.results.append(result)
+            args = [result]  # The output of one flow becomes the input to the next
+
+
+class Workflow:
+    def __init__(self):
+        self.tasks: Dict[str, Task] = {}
+        self.executor = concurrent.futures.ThreadPoolExecutor()
+
+    def add_task(self, task: Task):
+        self.tasks[task.id] = task
+
+    def run(self):
+        completed_tasks = set()
+        while len(completed_tasks) < len(self.tasks):
+            futures = []
+            for task in self.tasks.values():
+                if task.id not in completed_tasks and all(
+                    dep in completed_tasks for dep in task.dependencies
+                ):
+                    future = self.executor.submit(
+                        task.execute,
+                        {dep: self.tasks[dep].results for dep in task.dependencies},
+                    )
+                    futures.append((future, task.id))
+
+            for future, task_id in futures:
+                future.result()  # Wait for task completion
+                completed_tasks.add(task_id)
+
+    def get_results(self):
+        return {task_id: task.results for task_id, task in self.tasks.items()}
+
+
+# create flows
+llm = OpenAIChat(openai_api_key="sk-")
+
+flow1 = Flow(llm, max_loops=1)
+flow2 = Flow(llm, max_loops=1)
+flow3 = Flow(llm, max_loops=1)
+flow4 = Flow(llm, max_loops=1)
+
+
+# Create tasks with their respective Flows and task strings
+task1 = Task("task1", "Generate a summary on Quantum field theory", [flow1])
+task2 = Task("task2", "Elaborate on the summary of topic X", [flow2, flow3], dependencies=["task1"])
+task3 = Task("task3", "Generate conclusions for topic X", [flow4], dependencies=["task1"])
+
+# Create a workflow and add tasks
+workflow = Workflow()
+workflow.add_task(task1)
+workflow.add_task(task2)
+workflow.add_task(task3)
+
+# Run the workflow
+workflow.run()
+
+# Get results
+results = workflow.get_results()
+print(results)

swarms/utils/main.py

@@ -387,4 +387,4 @@ def handle(self, url: str) -> str:
 
 # =>  base end
 
-# ===========================>
+# ===========================>