Workflow Diagram Generation#

Source: examples/workflow/workflow_diagram_generation.py

Introduction#

Workflow definitions stay readable while they are small, but nested loops and conditional routing get harder to inspect once orchestration grows. This example shows how to generate a deterministic Mermaid diagram directly from a configured Workflow so the same topology can be reused in local development, docs, and review discussions.

Technical Implementation#

  1. Build a representative workflow using only the public workflow primitives.

  2. Call Workflow.to_mermaid(direction="LR") to render the declared topology.

  3. Call Workflow.to_svg(direction="LR") to emit a static SVG for notebooks, docs assets, or reviews.

  4. Persist both diagram formats under artifacts/examples/ for local inspection or docs reuse.

  5. Print a compact JSON payload so example automation can verify the generated diagram shape.

The diagram below is generated from the example’s configured Workflow.

        flowchart LR
    workflow_entry["Workflow Entrypoint"]
    step_1["prepare<br/>LogicStep"]
    step_2["review_loop<br/>LoopStep<br/>max_iterations=2"]
    step_3["publish<br/>LogicStep"]
    subgraph loop_body_1["Loop Body: review_loop"]
        direction TD
        loop_entry_1["review_loop iteration entry"]
        step_4["review_loop::router<br/>LogicStep"]
        step_5["review_loop::draft<br/>LogicStep"]
        step_6["review_loop::score<br/>LogicStep"]
        loop_entry_1 --> step_4
        step_4 -. "route=draft_path" .-> step_5
        step_4 -. "route=score_path" .-> step_6
        step_5 -. "next iteration" .-> loop_entry_1
        step_6 -. "next iteration" .-> loop_entry_1
    end
    workflow_entry --> step_1
    step_1 --> step_2
    step_2 -. "iterate" .-> loop_entry_1
    step_2 --> step_3
    
 1from __future__ import annotations
 2
 3import json
 4from pathlib import Path
 5
 6import design_research_agents as drag
 7
 8WORKFLOW_DIAGRAM_DIRECTION = "LR"
 9
10
11def build_example_workflow() -> drag.Workflow:
12    """Build a representative workflow with routing and loop structure."""
13    return drag.Workflow(
14        steps=[
15            drag.LogicStep(
16                step_id="prepare",
17                handler=lambda _context: {"prompt_ready": True},
18            ),
19            drag.LoopStep(
20                step_id="review_loop",
21                dependencies=("prepare",),
22                max_iterations=2,
23                steps=(
24                    drag.LogicStep(
25                        step_id="router",
26                        handler=lambda _context: {"route": "draft_path"},
27                        route_map={
28                            "draft_path": ("draft",),
29                            "score_path": ("score",),
30                        },
31                    ),
32                    drag.LogicStep(
33                        step_id="draft",
34                        dependencies=("router",),
35                        handler=lambda _context: {"draft": "candidate"},
36                    ),
37                    drag.LogicStep(
38                        step_id="score",
39                        dependencies=("router",),
40                        handler=lambda _context: {"score": 0.9},
41                    ),
42                ),
43            ),
44            drag.LogicStep(
45                step_id="publish",
46                dependencies=("review_loop",),
47                handler=lambda _context: {"published": True},
48            ),
49        ]
50    )
51
52
53def main() -> None:
54    """Generate Mermaid and SVG output for a representative workflow."""
55    workflow = build_example_workflow()
56    diagram = workflow.to_mermaid(direction=WORKFLOW_DIAGRAM_DIRECTION)
57    svg = workflow.to_svg(direction=WORKFLOW_DIAGRAM_DIRECTION)
58
59    mermaid_path = Path("artifacts/examples/workflow_diagram.mmd")
60    svg_path = Path("artifacts/examples/workflow_diagram.svg")
61    mermaid_path.parent.mkdir(parents=True, exist_ok=True)
62    mermaid_path.write_text(diagram + "\n", encoding="utf-8")
63    svg_path.write_text(svg + "\n", encoding="utf-8")
64
65    print(
66        json.dumps(
67            {
68                "diagram_path": mermaid_path.as_posix(),
69                "svg_path": svg_path.as_posix(),
70                "line_count": len(diagram.splitlines()),
71                "starts_with": diagram.splitlines()[0],
72                "svg_starts_with": svg.lstrip()[:4],
73                "contains_loop": "Loop Body: review_loop" in diagram,
74                "contains_route": "route=draft_path" in diagram,
75            },
76            ensure_ascii=True,
77            indent=2,
78            sort_keys=True,
79        )
80    )
81
82
83if __name__ == "__main__":
84    main()

Expected Results#

Run Command

PYTHONPATH=src python3 examples/workflow/workflow_diagram_generation.py

Example output shape (values vary by run):

{
  "diagram_path": "artifacts/examples/workflow_diagram.mmd",
  "svg_path": "artifacts/examples/workflow_diagram.svg",
  "line_count": 21,
  "starts_with": "flowchart LR",
  "svg_starts_with": "<svg",
  "contains_loop": true,
  "contains_route": true
}

References#