Ontological validation in pyiron_workflow

In pyiron_workflow, we leverage pyiron’s ontological validation of workflow graphs found in semantikon. We strive to support every syntactic aspect of semantikon, which you can read more about on the GitHub repository.

This is accomplished practically by wrapping function IO (or dataclass nodes) with the semantikon.metadata.u call. Here, we demonstrate a variety of use-cases, and show how this functionality can be taken further to get data type- and ontologically-valid suggestions for new connections in your workflow, or new nodes to add to it.

Note that ontological typing only works when there’s a parent object around (a macro or a workflow).

Ontological connection checking

Some success and failure cases in the presence or absence of ontological hints. A key takeway is that these hints function very much like pyiron_workflow’s regular type hinting: if hints are present on both sides of a new connection, they must be valid, but if one or both sides are missing the hint we skip the validation.

import rdflib

from semantikon.metadata import u, SemantikonURI

import pyiron_workflow as pwf
from pyiron_workflow import suggest
from pyiron_workflow.channels import ChannelConnectionError
from pyiron_workflow.knowledge import validate_workflow
from pyiron_workflow.nodes.composite import FailedChildError


EX = rdflib.Namespace("http://www.example.org/")

class Meal: ...

class Garbage: ...

@pwf.as_function_node("pizza")
def prepare_pizza() -> u(Meal, uri=EX.Pizza):
    return Meal()

@pwf.as_function_node("unidentified_meal")
def prepare_non_ontological_meal() -> Meal:
    return Meal()

@pwf.as_function_node("rice")
def prepare_rice() -> u(Meal, uri=EX.Rice):
    return Meal()

@pwf.as_function_node("garbage")
def prepare_garbage() -> u(Garbage, uri=EX.Garbage):
    return Garbage()

@pwf.as_function_node("garbage")
def prepare_unhinted_garbage():
    return Garbage()

@pwf.as_function_node("verdict")
def eat(meal: u(Meal, uri=EX.Meal)) -> str:
    return f"Yummy {meal.__class__.__name__} meal"

@pwf.as_function_node("verdict")
def eat_pizza(meal: u(Meal, uri=EX.Pizza)) -> str:
    return f"Yummy {meal.__class__.__name__} pizza"

Both fully hinted

Works fine

wf = pwf.Workflow("ontoflow")
wf.make = prepare_pizza()
wf.eat = eat_pizza(wf.make)
wf()

{'eat__verdict': 'Yummy Meal pizza'}

Upstream type hint is missing

Standard pyiron_workflow typing behaviour: we are allowed to form the connection (since the source has no hint), but at runtime, we will fail when we try to actually assign the value

wf = pwf.Workflow("no_type")
wf.make = prepare_unhinted_garbage()
wf.eat = eat_pizza(wf.make)
try:
    wf.recovery = None
    wf()
except FailedChildError as e:
    print(e)

/no_type encountered error in child: {'/no_type/eat.accumulate_and_run': TypeError("The channel /no_type/eat.meal cannot take the value `<__main__.Garbage object at 0x12a6ae600>` (<class '__main__.Garbage'>) because it is not compliant with the type hint typing.Annotated[__main__.Meal, ('uri', rdflib.term.URIRef('http://www.example.org/Pizza'))]")}

Upstream type hint is wrong

Standard pyiron_workflow typing behaviour: we’re not even allowed to form the connection – the recipe would be invalid

wf = pwf.Workflow("no_type")
wf.make = prepare_garbage()
try:
    wf.eat = eat_pizza(wf.make)
except ChannelConnectionError as e:
    print(e)

The upstream channel /no_type/make.garbage cannot connect to the downstream channel /no_type/eat_pizza.meal because the upstream type hint (typing.Annotated[__main__.Garbage, ('uri', rdflib.term.URIRef('http://www.example.org/Garbage'))]) is not as or more specific than the downstream type hint (typing.Annotated[__main__.Meal, ('uri', rdflib.term.URIRef('http://www.example.org/Pizza'))]).

So far, so good: u decoration has no negative impact on the existing type hint checking procedures

Upstream ontological hint is missing

New ontological behaviour: As with type hints, if one side is missing we just let things pass. Unlike type hints, we can also execute the workflow, because the ontologies only impact the recipe-level behaviour, not the instance behaviour!

wf = pwf.Workflow("no_ontology")
wf.make = prepare_non_ontological_meal()
wf.eat = eat_pizza(wf.make)
wf()

{'eat__verdict': 'Yummy Meal pizza'}

Upstream ontological hint is WRONG

New ontological behaviour: new ontological type checking now prevents us from even forming the ontologically invalid connection!

wf = pwf.Workflow("failed_ontology")
wf.make = prepare_rice()
try:
    wf.eat = eat_pizza(wf.make)
except ChannelConnectionError as e:
    print(e)

The upstream channel /failed_ontology/make.rice cannot connect to the downstream channel /failed_ontology/eat_pizza.meal because the upstream type hint (typing.Annotated[__main__.Meal, ('uri', rdflib.term.URIRef('http://www.example.org/Rice'))]) and downstream type hint (typing.Annotated[__main__.Meal, ('uri', rdflib.term.URIRef('http://www.example.org/Pizza'))]) produce a non-empty ontological validation report:
(False, <Graph identifier=Nd1ec45099900420db586c5ec9f925fe6 (<class 'rdflib.graph.Graph'>)>, 'Validation Report\nConforms: False\nResults (1):\nConstraint Violation in ClassConstraintComponent (http://www.w3.org/ns/shacl#ClassConstraintComponent):\n\tSeverity: sh:Violation\n\tSource Shape: [ rdf:type sh:PropertyShape ; sh:class <http://www.example.org/Pizza> ; sh:path <http://purl.obolibrary.org/obo/OBI_0001927> ]\n\tFocus Node: sns:c30f63af454c39d79ea4a3b85ab5da83_failed_ontology-make-outputs-rice_data\n\tValue Node: sns:c30f63af454c39d79ea4a3b85ab5da83_failed_ontology-make-outputs-rice_data_uri\n\tResult Path: <http://purl.obolibrary.org/obo/OBI_0001927>\n\tMessage: Value does not have class <http://www.example.org/Pizza>\n')

Downstream ontological hint is less specific

This should work fine…

wf = pwf.Workflow("relaxed_ontology")
wf.make = prepare_rice()
try:
    wf.eat = eat(wf.make)
except ChannelConnectionError as e:
    print(e)

The upstream channel /relaxed_ontology/make.rice cannot connect to the downstream channel /relaxed_ontology/eat.meal because the upstream type hint (typing.Annotated[__main__.Meal, ('uri', rdflib.term.URIRef('http://www.example.org/Rice'))]) and downstream type hint (typing.Annotated[__main__.Meal, ('uri', rdflib.term.URIRef('http://www.example.org/Meal'))]) produce a non-empty ontological validation report:
(False, <Graph identifier=Ndd35d574c31048baac18f7a6024d5a3b (<class 'rdflib.graph.Graph'>)>, 'Validation Report\nConforms: False\nResults (1):\nConstraint Violation in ClassConstraintComponent (http://www.w3.org/ns/shacl#ClassConstraintComponent):\n\tSeverity: sh:Violation\n\tSource Shape: [ rdf:type sh:PropertyShape ; sh:class <http://www.example.org/Meal> ; sh:path <http://purl.obolibrary.org/obo/OBI_0001927> ]\n\tFocus Node: sns:1f1e624b8164164e2ecce02fdf5532b5_relaxed_ontology-make-outputs-rice_data\n\tValue Node: sns:1f1e624b8164164e2ecce02fdf5532b5_relaxed_ontology-make-outputs-rice_data_uri\n\tResult Path: <http://purl.obolibrary.org/obo/OBI_0001927>\n\tMessage: Value does not have class <http://www.example.org/Meal>\n')

But! We forgot something! This form of failure is known from the semantikon notebook whence these demonstration workflow spring: we never informed the ontology that “rice” is a subclass of “meal”!

We let the ontology know this by adding the corresponding triple to our rdflib.Graph. In pyiron_workflow we can manage this by pre-populating a knowledge: rdflib.Graph property on the graph root (i.e. top-most object) as follows:

wf = pwf.Workflow("relaxed_ontology")

wf.knowledge = rdflib.Graph()
wf.knowledge.add((EX.Rice, rdflib.RDFS.subClassOf, EX.Meal))

wf.make = prepare_rice()
wf.eat = eat(wf.make)
wf()

{'eat__verdict': 'Yummy Meal meal'}

Ontological triples

Alright, for our simple pizza example things are working beautifully. Let’s try it with the clothes example. For output triples, we leverage the dual A-/T-box SemantikonURI wrapper.

EX = rdflib.Namespace("http://www.example.org/")

uri_cleaned = SemantikonURI(EX.cleaned)
uri_color = SemantikonURI(EX.color)

class Clothes:
    pass

@pwf.as_function_node
def wash(clothes: u(Clothes, uri=EX.Clothes)) -> u(
    Clothes,
    uri=EX.Clothes,
    triples=(EX.hasProperty, uri_cleaned),
    derived_from="inputs.clothes"
):
    ...
    return clothes

@pwf.as_function_node
def dye(clothes: u(Clothes, uri=EX.Clothes), color="blue") -> u(
    Clothes,
    uri=EX.Clothes,
    triples=(EX.hasProperty, uri_color),
    derived_from="inputs.clothes",
):
    ...
    return clothes

@pwf.as_function_node
def sell(
    clothes: u(
        Clothes,
        uri=EX.Clothes,
        restrictions=(
            ((rdflib.OWL.onProperty, EX.hasProperty), (rdflib.OWL.someValuesFrom, EX.cleaned)),
            ((rdflib.OWL.onProperty, EX.hasProperty), (rdflib.OWL.someValuesFrom, EX.color)),
        )
    )
) -> int:
    price = 10
    return price

Now with restrictions

In the base case, everything works fine. The restrictions are correctly parsed.

Note that unlike the semantikon notebook, here we had to make sure that all the node inputs are also u annotated (even if it’s just to trivially link the type to its ontology counterpart). This is because type checking only occurs in pyiron_workflow when both sides of the connection are typed! We follow this rule for both standard data types and ontological types.

my_correct_wf = pwf.Workflow("my_correct_workflow")
my_correct_wf.dyed_clothes = dye(Clothes())
my_correct_wf.washed_clothes = wash(my_correct_wf.dyed_clothes)
my_correct_wf.money = sell(my_correct_wf.washed_clothes)
my_correct_wf()

{'money__price': 10}

As a macro

This also works fine! Be careful though, here we’ve only demonstrated that it can work for macros, and have not yet guaranteed it works for all macros.

@pwf.as_macro_node
def my_correct_macro(self, clothes: Clothes):
    self.dyed_clothes = dye(clothes)
    self.washed_clothes = wash(self.dyed_clothes)
    self.money = sell(self.washed_clothes)
    return self.money

correct_m = my_correct_macro(Clothes())
correct_m()

{'money': 10}

Trivial failure

If we skip a step, our sell restrictions are not fulfilled, and we sensibly fail.

my_wrong_wf = pwf.Workflow("my_wrong_workflow")
my_wrong_wf.washed_clothes = wash(Clothes())
try:
    my_wrong_wf.money = sell(my_wrong_wf.washed_clothes)
except ChannelConnectionError as e:
    print(e)

The upstream channel /my_wrong_workflow/washed_clothes.clothes cannot connect to the downstream channel /my_wrong_workflow/sell.clothes because the upstream type hint (typing.Annotated[__main__.Clothes, ('uri', rdflib.term.URIRef('http://www.example.org/Clothes'), 'triples', (rdflib.term.URIRef('http://www.example.org/hasProperty'), SemantikonURI('http://www.example.org/cleaned')), 'derived_from', 'inputs.clothes')]) and downstream type hint (typing.Annotated[__main__.Clothes, ('uri', rdflib.term.URIRef('http://www.example.org/Clothes'), 'restrictions', (((rdflib.term.URIRef('http://www.w3.org/2002/07/owl#onProperty'), rdflib.term.URIRef('http://www.example.org/hasProperty')), (rdflib.term.URIRef('http://www.w3.org/2002/07/owl#someValuesFrom'), rdflib.term.URIRef('http://www.example.org/cleaned'))), ((rdflib.term.URIRef('http://www.w3.org/2002/07/owl#onProperty'), rdflib.term.URIRef('http://www.example.org/hasProperty')), (rdflib.term.URIRef('http://www.w3.org/2002/07/owl#someValuesFrom'), rdflib.term.URIRef('http://www.example.org/color')))))]) produce a non-empty ontological validation report:
(False, <Graph identifier=N56dfb7e84d254992b66430e8925b6c6c (<class 'rdflib.graph.Graph'>)>, 'Validation Report\nConforms: False\nResults (1):\nConstraint Violation in QualifiedValueShapeConstraintComponent (http://www.w3.org/ns/shacl#QualifiedMinCountConstraintComponent):\n\tSeverity: sh:Violation\n\tSource Shape: [ rdf:type sh:PropertyShape ; sh:path <http://www.example.org/hasProperty> ; sh:qualifiedMinCount Literal("1", datatype=xsd:integer) ; sh:qualifiedValueShape [ sh:class <http://www.example.org/color> ] ]\n\tFocus Node: sns:57c60cb96739908152ad395c29fb76e4_my_wrong_workflow-washed_clothes-outputs-clothes_data\n\tResult Path: <http://www.example.org/hasProperty>\n\tMessage: Focus node does not conform to shape MinCount 1: [ sh:class <http://www.example.org/color> ]\n')

Macro failure

When we wrap the failing code as a macro, we don’t fail until we try to instantiate that macro – that is the first time the recipe code is evaluated and ontologically evaluated, at which point we fail at the connection formation just like in the workflow example.

In the future, if we move to pyiron_workflow decorators first producing (and validating) flowrep recipes and then using these to create pyiron_workflow node classes, we’d be able to nicely fail at the macro definition time instead!

@pwf.as_macro_node
def my_wrong_macro(self, clothes: Clothes):
    self.washed_clothes = wash(clothes)
    self.money = sell(self.washed_clothes)
    return self.money

try:
    my_wrong_macro()
except ChannelConnectionError as e:
    print(e)

The upstream channel /my_wrong_macro/washed_clothes.clothes cannot connect to the downstream channel /my_wrong_macro/sell.clothes because the upstream type hint (typing.Annotated[__main__.Clothes, ('uri', rdflib.term.URIRef('http://www.example.org/Clothes'), 'triples', (rdflib.term.URIRef('http://www.example.org/hasProperty'), SemantikonURI('http://www.example.org/cleaned')), 'derived_from', 'inputs.clothes')]) and downstream type hint (typing.Annotated[__main__.Clothes, ('uri', rdflib.term.URIRef('http://www.example.org/Clothes'), 'restrictions', (((rdflib.term.URIRef('http://www.w3.org/2002/07/owl#onProperty'), rdflib.term.URIRef('http://www.example.org/hasProperty')), (rdflib.term.URIRef('http://www.w3.org/2002/07/owl#someValuesFrom'), rdflib.term.URIRef('http://www.example.org/cleaned'))), ((rdflib.term.URIRef('http://www.w3.org/2002/07/owl#onProperty'), rdflib.term.URIRef('http://www.example.org/hasProperty')), (rdflib.term.URIRef('http://www.w3.org/2002/07/owl#someValuesFrom'), rdflib.term.URIRef('http://www.example.org/color')))))]) produce a non-empty ontological validation report:
(False, <Graph identifier=Nd91c46ed4484401b813652723239465f (<class 'rdflib.graph.Graph'>)>, 'Validation Report\nConforms: False\nResults (1):\nConstraint Violation in QualifiedValueShapeConstraintComponent (http://www.w3.org/ns/shacl#QualifiedMinCountConstraintComponent):\n\tSeverity: sh:Violation\n\tSource Shape: [ rdf:type sh:PropertyShape ; sh:path <http://www.example.org/hasProperty> ; sh:qualifiedMinCount Literal("1", datatype=xsd:integer) ; sh:qualifiedValueShape [ sh:class <http://www.example.org/color> ] ]\n\tFocus Node: sns:8ff9ae536d3445386e46abebb053590c_my_wrong_macro-washed_clothes-outputs-clothes_data\n\tResult Path: <http://www.example.org/hasProperty>\n\tMessage: Focus node does not conform to shape MinCount 1: [ sh:class <http://www.example.org/color> ]\n')

Ruling things out with knowledge

In our “meal” example, we saw that peer-to-peer edges must have commensurate URIs, and we used the workflow’s knowledge attribute to indicate a subclass relationship.

For parent-child edges negotiating subgraph IO, semantikon instead uses an open-world reasoning approach, such that URIs are not required to have a pre-existing relationship. For example, we can write a macro with our wash node, which both expects and returns a uri=EX.Clothes object, and interface with it using different URIs. This validates fine:

@pwf.as_macro_node
def wash_something_macro(self, something_to_wash: u(Clothes, uri=EX.MaybeClothesMaybeNot)) -> u(Clothes, EX.SomethingGotWashed):
    self.washed_something = wash(something_to_wash)
    return self.washed_something

wf = wash_something_macro()
print("Valid: ", validate_workflow(wf)[0])

Valid:  True

If we want, we can preclude such a graph from validating by providing explicit disjointness as contextual knowledge. This works for both the input and output edges independently, and is accomplished by the OWL disjointWith predicate:

external_knowledge = rdflib.Graph()
external_knowledge.add((EX.MaybeClothesMaybeNot, rdflib.OWL.disjointWith, EX.Clothes))
external_knowledge.add((EX.MaybeClothesMaybeNot, rdflib.RDF.type, rdflib.OWL.Class))
external_knowledge.add((EX.Clothes, rdflib.RDF.type, rdflib.OWL.Class))

print("Valid with input disjointness: ", validate_workflow(wf, knowledge=external_knowledge)[0])

external_knowledge = rdflib.Graph()
external_knowledge.add((EX.SomethingGotWashed, rdflib.OWL.disjointWith, EX.Clothes))
external_knowledge.add((EX.SomethingGotWashed, rdflib.RDF.type, rdflib.OWL.Class))
external_knowledge.add((EX.Clothes, rdflib.RDF.type, rdflib.OWL.Class))

print("Valid with output disjointness: ", validate_workflow(wf, knowledge=external_knowledge)[0])

Valid with input disjointness:  False
Valid with output disjointness:  False

Node suggestions

One of the advantages of graph-based workflows with hinted IO channels is facilitating guided workflow creation. Given a hinted channel instance in the context of some workflow, we can ask for suggestions of other channels with which to form a connection in the same, sibling graph context:

wf = pwf.Workflow("ontoflow")
wf.make = prepare_pizza()
wf.eat = eat_pizza()
suggestions = suggest.suggest_connections(wf.eat.inputs.meal)
for (node, channel) in suggestions:
    print(node.full_label, channel.label)

/ontoflow/make pizza

Similarly, given a corpus of node classes, we can ask for which nodes have at least one commensurate input/output with which our channel might connect. After adding such a node to our graph, we can leverage the connection suggester to see which channel(s) are appropriate.

suggest.suggest_nodes(wf.eat.inputs.meal, pwf.std.UserInput, prepare_pizza, wash)

[__main__.prepare_pizza]

Suggestion limitations

When searching for new upstream nodes to add, the current implementation only looks at the immediate node, and not possible trees of upstream nodes. Returning to our clothes example, we can see that there is no single suggestion for the sell node, because it requires clothes that are both dyed and coloured, but our other nodes only provide one of these at a time!

clothing_nodes = wash, dye, sell

wf = pwf.Workflow("working_backwards")
wf.money = sell()
suggest.suggest_nodes(wf.money, *clothing_nodes)

[]

Of course working backwards a single step still works fine for lots of nodes, e.g. for dye we will take anything that gives us clothes!

wf = pwf.Workflow("single_step_back")
wf.dyed_clothes = dye()
suggest.suggest_nodes(wf.dyed_clothes, *clothing_nodes)

[__main__.wash, __main__.dye]

And when we look downstream we have the advantage of knowing the entire upstream graph concretely, so there we are able to see options for fulfilling these more complex demands.

wf = pwf.Workflow("downstream")
wf.dyed_clothes = dye(Clothes())
wf.washed_clothes = wash(wf.dyed_clothes)
suggestions = suggest.suggest_nodes(wf.washed_clothes, *clothing_nodes)
assert(sell in suggestions)
print(suggestions)

[<class '__main__.wash'>, <class '__main__.dye'>, <class '__main__.sell'>]

When looking whether requirements are fulfilled, an input only cares that it’s own requirements are fulfilled; we may suggest upstream sources that wind up adding new restrictions on our terminal input. For instance, below TakesDownstream only cares that GivesAndTakes promises to supply something with the “Downstream” characteristic – it doesn’t mind that the GivesAndTakes input from which this output derive now further demands an “Upstream” characteristic:

uri_Upstream = SemantikonURI(EX.Upstream)
uri_Downstream = SemantikonURI(EX.Downstream)

@pwf.as_function_node
def GivesUpstreamNeed(x) -> u(str, uri=EX.Data, triples=(EX.has, uri_Upstream)):
    return str(x)

@pwf.as_function_node
def GivesAndTakes(
    y: u(
        str,
        uri=EX.Data,
        restrictions=(
            (rdflib.OWL.onProperty, EX.has),
            (rdflib.OWL.someValuesFrom, EX.Upstream),
        ),
    ),
) -> u(
    str,
    uri=EX.Data,
    derived_from="inputs.y",
    triples=(EX.has, uri_Downstream)
):
    return y

@pwf.as_function_node
def TakesDownstream(
    z: u(
        str,
        uri=EX.Data,
        restrictions=(
            (rdflib.OWL.onProperty, EX.has),
            (rdflib.OWL.someValuesFrom, EX.Downstream),
        ),
    ),
) -> str:
    return z

wf = pwf.Workflow("derived_restrictions")
wf.up = GivesUpstreamNeed()
wf.middle = GivesAndTakes()
wf.down = TakesDownstream()

suggest.suggest_connections(wf.middle.outputs.y)

[(<__main__.TakesDownstream at 0x13b3d6ff0>,
  <pyiron_workflow.channels.InputData at 0x12a2aae70>)]

Complex workflows

Ontological validation is still a new feature, and you may find an edge case we haven’t found and tested yet. In general, atomic function nodes and workflows/macro nodes should play nicely. Flow control nodes (for, while, etc.) and other node types are not uniformly supported and your milage may vary.

Units

semantikon annotations also allow us to specify physical units. When present, these are included in the ontological validation just like the other ontological terms.

As such, we have no problem making same-unit connections:

@pwf.as_function_node
def Distance(x: u(float, units="meter")) -> u(float, derived_from="inputs.x"):
    return x

@pwf.as_function_node
def Speed(
        dx: u(float, units="meter"), dt: u(float, units="second")
) -> u(float, units="meter/second"):
    s = dx/dt
    return s

wf = pwf.Workflow("speedometer")
wf.dx = Distance(100)
wf.speed = Speed(dx=wf.dx)

With incompatible units, we get an exception at connection time, just like with other ontological failures:

@pwf.as_function_node
def NanoTime(t: u(float, units="nanosecond")) -> u(float, units="nanosecond"):
    return t

wf.dt = NanoTime(10)
try:
    wf.speed.inputs.dt = wf.dt
except ChannelConnectionError as e:
    print(e)
    wf.remove_child(wf.dt)

The upstream channel /speedometer/dt.t cannot connect to the downstream channel /speedometer/speed.dt because the upstream type hint (typing.Annotated[float, ('units', 'nanosecond')]) and downstream type hint (typing.Annotated[float, ('units', 'second')]) produce a non-empty ontological validation report:
(False, <Graph identifier=N05096895db3d4125a01ca7c9c705d85e (<class 'rdflib.graph.Graph'>)>, "Validation Report\nConforms: False\nResults (1):\nConstraint Violation in HasValueConstraintComponent (http://www.w3.org/ns/shacl#HasValueConstraintComponent):\n\tSeverity: sh:Violation\n\tSource Shape: [ rdf:type sh:PropertyShape ; sh:hasValue unit:SEC ; sh:path qudt:hasUnit ]\n\tFocus Node: sns:f997e651f109385915d4a770a7207d30_speedometer-dt-outputs-t_data\n\tResult Path: qudt:hasUnit\n\tMessage: Node sns:f997e651f109385915d4a770a7207d30_speedometer-dt-outputs-t_data->qudt:hasUnit does not contain a value in the set: ['unit:SEC']\n")

With correct units, it works fine

@pwf.as_function_node
def Time(t: u(float, units="second")) -> u(float, units="second"):
    return t

wf.dt = Time(10)
wf.speed.inputs.dt = wf.dt
wf()

{'speed__s': 10.0}

(Note that units are NOT inherited using the derived_from= flag – if two nodes use different units, you’ll need to add an explicit unit conversion node into your graph.)