first version of Workflow::to_mermaid method

maestro/src/mermaid.py

@@ -0,0 +1,99 @@
+#!/usr/bin/env python3
+
+# Copyright © 2025 IBM
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+class Mermaid:
+    # kind: sequenceDiagram or flowchart
+    # orientation: TD (top down), 
+    #              RL (right left) 
+    #              when kind is flowchart    
+    def __init__(self, workflow, kind="sequenceDiagram", orientation="TD"):
+        self.workflow = workflow[0]
+        self.kind = kind
+        self.orientation = orientation
+
+    # generates a mermaid markdown representation of the workflow
+    def to_markdown(self) -> str:
+        sb, markdown = "", ""
+        if self.kind == "sequenceDiagram":
+            markdown = self.__to_sequenceDiagram(sb)
+        elif self.kind == "flowchart":
+            markdown = self.__to_flowchart(sb, self.orientation)
+        else:
+            raise RuntimeError(f"Invalid Mermaid kind: {kind}")
+        return markdown
+
+    # private methods
+
+    # returns a markdown of the workflow as a mermaid sequence diagram
+    # 
+    # sequenceDiagram
+    # participant agent1
+    # participant agent2
+    #
+    # agent1->>agent2: step1
+    # agent2->>agent3: step2
+    # agent2-->>agent1: step3
+    # agent1->>agent3: step4
+    #
+    # See mermaid sequence diagram documentation: 
+    # https://mermaid.js.org/syntax/block.html
+    def __to_sequenceDiagram(self, sb):
+        sb += "sequenceDiagram\n"
+
+        for agent in self.workflow['spec']['template']['agents']:
+            sb += f"participant {agent}\n"
+
+        steps, i = self.workflow['spec']['template']['steps'], 0
+        for step in steps:
+            agentL = step['agent']
+            agentR = None
+            if i < (len(steps) - 1):
+                agentR = steps[i+1]['agent']            
+            if agentR != None:
+                sb += f"{agentL}->>{agentR}: {step['name']}\n"
+            else:
+                sb += f"{agentL}->>{agentL}: {step['name']}\n"
+            i = i + 1
+
+        return sb
+
+    # returns a markdown of the workflow as a mermaid sequence diagram
+    # 
+    # flowchart LR
+    # agemt1-- step1 -->agent2
+    # agemt2-- step2 -->agent3
+    # agemt3-- step3 -->agent3
+    #
+    # See mermaid sequence diagram documentation: 
+    # https://mermaid.js.org/syntax/block.html
+    def __to_flowchart(self, sb, orientation):
+        sb += f"flowchart {orientation}\n"
+        steps, i = self.workflow['spec']['template']['steps'], 0
+
+        for step in steps:
+            agentL = step['agent']
+            agentR = None
+            if i < (len(steps) - 1):
+                agentR = steps[i+1]['agent']            
+            if agentR != None:
+                sb += f"{agentL}-- {step['name']} -->{agentR}\n"
+            else:
+                sb += f"{agentL}-- {step['name']} -->{agentL}\n"
+            i = i + 1
+
+        return sb
+
+

maestro/src/workflow.py

@@ -16,6 +16,8 @@
 
 import os, dotenv
 
+from src.mermaid import Mermaid
+
 from src.step import Step
 from src.agents.agent_factory import AgentFramework
 
@@ -56,6 +58,12 @@ def __init__(self, agent_defs={}, workflow={}):
         self.agent_defs = agent_defs
         self.workflow = workflow
 
+    # generates a mermaid markdown representation of the workflow
+    # kind: sequenceDiagram or flowchart
+    # orientation: TD (top down) or RL (right left) when kind is flowchart
+    def to_mermaid(self, kind="sequenceDiagram", orientation="TD") -> str:
+        return Mermaid(self.workflow, kind, orientation).to_markdown()
+
     async def run(self):
         """Execute workflow."""
         self.create_or_restore_agents(self.agent_defs, self.workflow)

maestro/tests/workflow/test_mermaid.py

@@ -0,0 +1,97 @@
+#!/usr/bin/env python3
+
+# Copyright © 2025 IBM
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os, yaml
+
+from unittest import TestCase
+
+from src.mermaid import Mermaid
+from src.workflow import Workflow
+
+def parse_yaml(file_path):
+    with open(file_path, "r") as file:
+        yaml_data = list(yaml.safe_load_all(file))
+    return yaml_data
+
+class TestMermaid(TestCase):
+    def setUp(self):        
+        self.workflow_yaml = parse_yaml(os.path.join(os.path.dirname(__file__),"../yamls/workflows/simple_workflow.yaml"))
+
+    def tearDown(self):
+        self.workflow_yaml = None
+
+    def test_markdown_sequenceDiagram(self):
+        mermaid = Mermaid(self.workflow_yaml, "sequenceDiagram")
+        self.assertTrue(mermaid.to_markdown().startswith("sequenceDiagram"))
+
+        for agent in ['test1', 'test2', 'test3']:
+            self.assertTrue(f"participant {agent}" in mermaid.to_markdown())
+
+        self.assertTrue(f"test1->>test2: step1" in mermaid.to_markdown())
+        self.assertTrue(f"test2->>test3: step2" in mermaid.to_markdown())
+        self.assertTrue(f"test3->>test3: step3" in mermaid.to_markdown())
+
+    def test_markdown_flowchartTD(self):
+        mermaid = Mermaid(self.workflow_yaml, "flowchart", "TD")
+        self.assertTrue(mermaid.to_markdown().startswith("flowchart TD"))
+
+        self.assertTrue(f"test1-- step1 -->test2" in mermaid.to_markdown())
+        self.assertTrue(f"test2-- step2 -->test3" in mermaid.to_markdown())
+        self.assertTrue(f"test3-- step3 -->test3" in mermaid.to_markdown())
+
+    def test_markdown_flowchartRL(self):
+        mermaid = Mermaid(self.workflow_yaml, "flowchart", "RL")
+        self.assertTrue(mermaid.to_markdown().startswith("flowchart RL"))
+
+        self.assertTrue(f"test1-- step1 -->test2" in mermaid.to_markdown())
+        self.assertTrue(f"test2-- step2 -->test3" in mermaid.to_markdown())
+        self.assertTrue(f"test3-- step3 -->test3" in mermaid.to_markdown())
+
+class TestWorkdlow_to_mermaid(TestCase):
+    def setUp(self):        
+        self.workflow_yaml = parse_yaml(os.path.join(os.path.dirname(__file__),"../yamls/workflows/simple_workflow.yaml"))
+        self.workflow = Workflow({}, self.workflow_yaml)
+
+    def tearDown(self):
+        self.workflow_yaml = None
+        self.workflow = None
+
+    def test_markdown_sequenceDiagram(self):
+        markdown = self.workflow.to_mermaid("sequenceDiagram")
+        self.assertTrue(markdown.startswith("sequenceDiagram"))
+
+        self.assertTrue(f"test1->>test2: step1" in markdown)
+        self.assertTrue(f"test2->>test3: step2" in markdown)
+        self.assertTrue(f"test3->>test3: step3" in markdown)
+
+    def test_markdown_flowchartTD(self):
+        markdown = self.workflow.to_mermaid("flowchart", "TD")
+        self.assertTrue(markdown.startswith("flowchart TD"))
+
+        self.assertTrue(f"test1-- step1 -->test2" in markdown)
+        self.assertTrue(f"test2-- step2 -->test3" in markdown)
+        self.assertTrue(f"test3-- step3 -->test3" in markdown)
+
+    def test_markdown_flowchartRL(self):
+        markdown = self.workflow.to_mermaid("flowchart", "RL")
+        self.assertTrue(markdown.startswith("flowchart RL"))
+
+        self.assertTrue(f"test1-- step1 -->test2" in markdown)
+        self.assertTrue(f"test2-- step2 -->test3" in markdown)
+        self.assertTrue(f"test3-- step3 -->test3" in markdown)
+
+if __name__ == '__main__':
+    unittest.main()

maestro/tests/yamls/workflows/simple_workflow.yaml

@@ -16,7 +16,7 @@ spec:
         - test2
         - test3
     prompt: This is a test input
-    exception: step4
+    exception: step3
     steps:
       - name: step1
         agent: test1