Common schemas for workflows

[ml-evs]

This thread can be used to record notes from general discussions in the working group on common workflow schemas.

Pre-workshop discussion: #16

[jan-janssen]

Summary

We created a repository for the results of our working group:
https://github.com/pythonworkflow/madices-workshop-2025

The initial paper for the Python Workflow Definition is finally available on Digital Discovery:
https://doi.org/10.1039/D5DD00231A

[schaarj]

reposting my version of the full list from #16:

we try to focus on following issues mostly centred around the Python Workflow definition (PWD)

Active Topics

• Adding while Loops to the PWD - moving away from the DAG
• Integrating PWD support into PerQueue
• WrapPWD in Dagster
• Implement a NOMAD Parser/Schema for the PWD
• Provide solutions for Remote Connections/Automated triggering of Workflows on remote Systems

Additional topics to be addressed if time permits

• Provide a Generic Interface/Exchange Format for data storage / Schema Mapping from data generated during a workflow run ()

Expected Deliverables at the end of the workshop

• new definition of PWD that supports dynamic workflows with while loops with examples
• PWD support in PerQueue
• PWD importer for SimStack
• NOMAD PWD Parser/Schema
• Example support for PWD in Dagster

[jan-janssen]

Current ToDos

• Check if we can if the AiiDA bug of functions with the same name: https://github.com/jan-janssen/pwd-recursive-while/blob/main/aiida_bug.ipynb
• Upgrade Python Workflow Definition to support latest version of aiida-workgraph Test with aiida-workgraph 0.7.4 pythonworkflow/python-workflow-definition#136
• Upgrade Python Workflow Definition to support latest version of pyiron_base pyiron_base update to 0.15.5 pythonworkflow/python-workflow-definition#135
• We noticed that we have to add the support for grouping, for example we could add a node for {"id": 0, "type": "workflow", "value": "workflow._group.json"},. This would be the nesting of a whole workflow in an external file.
• In addition, we could add the option to nest a workflow inside the same file.

[superstar54]

Check if we can if the AiiDA bug of functions with the same name: https://github.com/jan-janssen/pwd-recursive-while/blob/main/aiida_bug.ipynb

I tested this example using the latest 0.7.3, and there is no issue.

[jan-janssen]

It is working - thanks a lot for the quick fix.

[JaGeo]

Plan for extending the PWD for dynamic workflows:

• We will use the abstract syntax trees generated from pure Python for inspiration in designing how our PWD representation should look
• We will start with a simplified example with while loops, but we will consider further extensions in the future. We have collected examples how to implement workflows within aiida, pyiron and jobflow here: https://github.com/jan-janssen/pwd-recursive-while

[jan-janssen]

Draft of the JSON format

Inspiration from https://github.com/jan-janssen/while-loop-flowrep/blob/main/while.ipynb

Suggestion - for node four the output has to be x not result:

{
    "version": "0.1.1",
    "nodes": [
        {
            "id": 0, 
            "type": "while", 
            "value": {
                "nodes": [
                    {"id": 0, "type": "function", "value": "workflow.function_body"},
                    {"id": 1, "type": "test", "value": "workflow.condition"},
                    {"id": 2, "type": "input", "name": "x"},
                    {"id": 3, "type": "input", "name": "limit"},
                    {"id": 4, "type": "output", "name": "x"},
                ],
                "edges": [
                    {"target": 4, "targetPort": null, "source": 0, "sourcePort": null},
                    {"target": 0, "targetPort": "x", "source": 2, "sourcePort": null},
                    {"target": 1, "targetPort": "x", "source": 2, "sourcePort": null},
                    {"target": 1, "targetPort": "limit", "source": 3, "sourcePort": null},
                ]
            }
        },
        {"id": 1, "type": "input", "value": 0, "name": "x"},
        {"id": 2, "type": "input", "value": 5, "name": "limit"},
        {"id": 3, "type": "output", "name": "result"}
        
    ],
    "edges": [
        {"target": 0, "targetPort": "x", "source": 1, "sourcePort": null},
        {"target": 0, "targetPort": "limit", "source": 2, "sourcePort": null},
        {"target": 3, "targetPort": null, "source": 0, "sourcePort": "result"},
    ]
}

[giovannipizzi]

Before going into the discussion of the while loop, we discussed that we should support nesting of workflows written in JSON.

Here is a suggested format:

{
  "version": "0.2.0",
  "workflows": {
    "prod_div": {
      "nodes": [
        {"id": 0, "type": "function", "value": "workflow.get_prod_and_div"},
        {"id": 1, "type": "function", "value": "workflow.get_sum"},
        {"id": 2, "type": "function", "value": "workflow.get_square"},
        {"id": 3, "type": "input", "value": 1, "name": "x"},
        {"id": 4, "type": "input", "value": 2, "name": "y"},
        {"id": 5, "type": "output", "name": "result"}
      ],
      "edges": [
        {"target": 0, "targetPort": "x", "source": 3, "sourcePort": null},
        {"target": 0, "targetPort": "y", "source": 4, "sourcePort": null},
        {"target": 1, "targetPort": "x", "source": 0, "sourcePort": "prod"},
        {"target": 1, "targetPort": "y", "source": 0, "sourcePort": "div"},
        {"target": 2, "targetPort": "x", "source": 1, "sourcePort": null},
        {"target": 5, "targetPort": null, "source": 2, "sourcePort": null}
      ]
    },
    "main": {
      "nodes": [
          {"id": 0, "type": "workflow", "value": "prod_div"},
          {"id": 1, "value": 1, "type": "input", "name": "a"},
          {"id": 2, "value": 2, "type": "input", "name": "b"},
          {"id": 3, "type": "workflow", "value": "example.json:main"},
          {"id": 4, "type": "output", "name": "final_result"}
      ],
      "edges": [
          {"target": 0, "targetPort": "x", "source": 1, "sourcePort": null},
          {"target": 0, "targetPort": "y", "source": 2, "sourcePort": null},
          {"target": 3, "targetPort": "inp", "source": 0, "sourcePort": "result"},
          {"target": 4, "targetPort": null, "source": 3, "sourcePort": "out"}
      ]
    }
  }
}

Discussion:

• we want to have the option of defining a library of workflows. Therefore, we have a new top-level workflows key. Internally, the key is the workflow name (to call it from parent workflows), the content has the same schema as before in v 0.1.0, for which we also have pydantic models.
• in principle we could also keep a "main" workflow with edges and nodes outside. But in the end it's a bit strange to have this "special" workflow. So we thought to go to a new 0.2.0 version, and have everything as functions, and the "main" function that should be run when this JSON is executed be called "main".
• To call a sub workflow, you would now have a new type "workflow". If you pass a string that does not contain the colon character :, this should be a key in the workflows dictionary. Port names are defined by the input and output types in the sub workflow, and by their names. If you want to refer to a workflow in a different file, you can use the syntax filename.json:workflow_name, where the filename.json is the path to another JSON containing the workflow definitions, relative to the current file; and the part after the colon is the workflow name (inside the workflows key of that file).

This essentially allows for grouping of nodes into logical workflows, allows for arbitrary nesting levels for workflow calling, and to define libraries of common workflows.

[GeigerJ2]

We propose to be able to store only one workflow per file as it would reduce complexity of the format and it would be backwards compatible.

With this, we stay at 0.1.x and only add a workflow node type. Example above becomes this:

{
  "version": "0.1.1",
  "nodes": [
    { "id": 0, "type": "function", "value": "workflow.get_prod_and_div" },
    { "id": 1, "type": "function", "value": "workflow.get_sum" },
    { "id": 2, "type": "function", "value": "workflow.get_square" },
    { "id": 3, "type": "input", "value": 1, "name": "x" },
    { "id": 4, "type": "input", "value": 2, "name": "y" },
    { "id": 5, "type": "output", "name": "result" }
  ],
  "edges": [
    { "target": 0, "targetPort": "x", "source": 3, "sourcePort": null },
    { "target": 0, "targetPort": "y", "source": 4, "sourcePort": null },
    { "target": 1, "targetPort": "x", "source": 0, "sourcePort": "prod" },
    { "target": 1, "targetPort": "y", "source": 0, "sourcePort": "div" },
    { "target": 2, "targetPort": "x", "source": 1, "sourcePort": null },
    { "target": 5, "targetPort": null, "source": 2, "sourcePort": null }
  ]
}

{
  "version": "0.1.1",
  "nodes": [
    { "id": 0, "type": "workflow", "value": "prod_div.json" },
    { "id": 1, "value": 1, "type": "input", "name": "a" },
    { "id": 2, "value": 2, "type": "input", "name": "b" },
    { "id": 3, "type": "output", "name": "final_result" }
  ],
  "edges": [
    { "target": 0, "targetPort": "x", "source": 1, "sourcePort": null },
    { "target": 0, "targetPort": "y", "source": 2, "sourcePort": null },
    { "target": 3, "targetPort": "null", "source": 0, "sourcePort": "result" }
  ]
}

General guiding principles

• Only one representation of how something is done (interoperability format should be clearly well-defined)
• Minimal changes to previous version
• Minimal set of keywords

In the future: Validate that ports of the outer graph map correctly to the ports of the inner graph.

[JaGeo]

We also disussed calling the main file main.pwd.json, right?

[schaarj]

see my comment below but yes

[JaGeo]

I made a first PR to test how to implement this in jobflow.

While reading this initial example passes, more complex flows would not. Currently, the outputs from the nested flows are not correctly extracted anf moved on to an outer job or workflow.

pythonworkflow/python-workflow-definition#142

[jan-janssen]

For documentation I added some notebook to compare the nested workflows as they are implemented in the different workflow frameworks: https://github.com/jan-janssen/pwd-recursive-while/blob/main/day_4.ipynb

[JaGeo]

Example to generate nested flows

from jobflow import job, Flow, run_locally


@job
def add(x, y):
    return x + y


def get_flow(x,y):
    job1=add(x,y)
    job2=add(job1.output,y)
    return Flow([job1,job2],output=job2.output)

flow1=get_flow(1,2)
flow2=get_flow(flow1.output,4)

final_flow=Flow([flow1,flow2],output=flow2.output)

print(final_flow.as_dict())

run_locally(final_flow)

[schaarj]

in order to make the PWD files easily recognisable, we suggested to have a conventional name such as main.pwd.json and nested workflows as e.g. 'sub1.pwd.jsoen'

[giovannipizzi]

On a different topic, the while loop. Here is a possible implementation. However, the issue is that this might imply that if the body of the while loop is itself a workflow (written as a JSON file), to write it we write JSON (top level) calling python, calling JSON, etc etc.
This makes it complex, and also does not allow to parse the JSON only (no python) and understand the nested workflow structure.

Here is anyways the "pseudo" code, to stimulate discussions.
JSON PWD:

{
  "version": "0.2.0",
  "workflows": {
    "main": {
      "nodes": [
          {"id": 0, "value": "filename.DOS", "type": "input:from_file", "name": "DOS"},
          {"id": 1, "value": 0.0, "type": "input", "name": "EF_left"},
          {"id": 2, "value": 10.0, "type": "input", "name": "EF_right"},
          {"id": 3, "value": 0.001, "type": "input", "name": "convergence_threshold"},
          {"id": 4, "value": 1.0, "type": "input", "name": "target_num_electrons"},
          {"id": 5, "type": "while", "value": "function:body_f", "name": "while_loop", "condition_f": "condition_f", "initializer": "initializer_f", "finalizer": "finalizer_f"},
          {"id": 6, "type": "output", "name": "EF"},
          {"id": 7, "type": "output", "name": "num_electrons"}
      ],
      "edges": [
          {"source": 0, "sourcePort": null, "target": 5, "targetPort": "dos"},
          {"source": 1, "sourcePort": null, "target": 5, "targetPort": "EF_left"},
          {"source": 2, "sourcePort": null, "target": 5, "targetPort": "EF_right"},
          {"source": 3, "sourcePort": null, "target": 5, "targetPort": "target_num_electrons"},
          {"source": 4, "sourcePort": null, "target": 5, "targetPort": "convergence_threshold"},
          {"source": 5, "sourcePort": "EF", "target": 6, "targetPort": null},
          {"source": 5, "sourcePort": "num_electrons", "target": 7, "targetPort": null}
      ]
    }
  }
}

Python functions:

def initializer_f(input_ports, ctx):
    # Here you can do checks, or initialize other ctx variables.
    ctx['EF_left'] = input_ports['EF_left']
    ctx['EF_right'] = input_ports['EF_right']
    ctx['EF'] = (ctx['EF_left'] + ctx['EF_right']) / 2
    return ctx

def condition_f(input_ports, ctx):
    return (ctx['EF_right'] - ctx['EF_left']) > input_ports['convergence_threshold']

def body_f(input_ports, ctx):
    ctx['num_electrons'] = compute_num_electrons(input_ports['dos'], ctx['EF']) ### ISSUE: this should be another workflow! Could be a JSON, not a python function...
    if ctx['num_electrons'] < input_ports['target_num_electrons']:
        ctx['EF_left'] = ctx['EF']
    else:
        ctx['EF_right'] = ctx['EF']
    ctx['EF'] = (ctx['EF_left'] + ctx['EF_right']) / 2
    return ctx

def finalizer_f(input_ports, ctx):
    return {"EF": ctx['EF'], "num_electrons": ctx['num_electrons']}

[JFRudzinski]

Tomorrow I would request to briefly discuss my post in #29 whenever is convenient

[jan-janssen]

Together with @JFRudzinski I worked on a suggestion for an extension of the Python Workflow Definition which can include both the output Python objects as long as they are JSON serializable as well as the path to the directories where the node was executed. The second part can be used by frameworks like Nomad to parse the outputs with their own tools and integrate them in their database.
https://github.com/pythonworkflow/proposal-for-data-export

[PedeyTey]

Long overdue update on the PerQueue implementation of PWD

A PWD reader and writer has been developed in PerQueue, which can generate a functioning workflow within PerQueue. This also inspired explicit use of data ports for advanced data control, which currently works for most cases. It can read in the quantum espresso example from the python workflow definition and write this workflow again from the workflow generated from the read-in, and run a simplified version of this without explicit data porting. It can also submit this workflow with the option of generating temporary required files with submitted functions, if these do not exist already for easy workflow setup (note exception in point 2)*. For now, this submission will assume a local job submission with 5 minutes time limit, but this can be changed easily by the user.
Current challenges on this subject:

1. Implementing the PWD for the constructor tasks in PerQueue ("SwitchGroup": if-else construction; "WidthGroup": parallelisation of subgraph; "CyclicalGroup": while loop), and for nested workflows. In all these cases each subworkflow within the objects will be PDW writable and with data porting if specified, but the data transfer from outside any of these objects to the inside is assumed to fit the first task with no porting, and vice versa for the transfer out of the subworkflows.

2. *Properly translating edges on the form {"source" : t1, "sourcePort" : null, "target" : t2, "targetPort" : arg}, as I am yet to find a proper way to translate "null" to "all data" without assuming the number of outputs. Having both "sourcePort" and "targetPort" as null, will default to blindly transferring data and assume that the output of the source will fit with the input of the target.

On a side note, PerQueue now supports any function name, and will import it from a specified script, where it prior to this week would only support functions named "main" (which it will default to, if function is not specified for backwards compatibility).

Examples of these implementations have been placed in the perqueue/examples folder, and all of this can be found on the PWD_abstraction branch on the gitlab repository, https://gitlab.com/asm-dtu/perqueue.git

[edan-bainglass]

See #33 for an ongoing discussion on supporting PWD-driven workflows in FINALES.