Passive nodes. LynxScribe example.

server/basic_ops.py

@@ -52,12 +52,13 @@ def visualize_graph(graph: ops.Bundle, *, color_nodes_by: 'node_attribute' = Non
   return v
 
 @ops.op("View tables", view="table_view")
-def view_tables(dfs: ops.Bundle):
+def view_tables(bundle: ops.Bundle):
   v = {
     'dataframes': { name: {
       'columns': [str(c) for c in df.columns],
       'data': df.values.tolist(),
-    } for name, df in dfs.dfs.items() },
-    'relations': dfs.relations,
+    } for name, df in bundle.dfs.items() },
+    'relations': bundle.relations,
+    'other': bundle.other,
   }
   return v

server/lynxscribe_ops.py

@@ -0,0 +1,14 @@
+'''An example of passive ops. Just using LynxKite to describe the configuration of a complex system.'''
+from .ops import register_passive_op, Parameter as P
+
+register_passive_op('Scrape documents', inputs={}, params=[P('url', '')])
+register_passive_op('Extract graph')
+register_passive_op('Compute embeddings')
+register_passive_op('Vector DB', params=[P('backend', 'FAISS')])
+register_passive_op('Chat UI', outputs={})
+register_passive_op('Chat backend', inputs={})
+register_passive_op('WhatsApp', inputs={})
+register_passive_op('PII removal')
+register_passive_op('Intent classification')
+register_passive_op('System prompt', inputs={}, params=[P('prompt', 'You are a heplful chatbot.')])
+register_passive_op('LLM', params=[P('model', 'gpt4')])

server/main.py

@@ -53,6 +53,10 @@ def get_catalog():
 def execute(ws):
     # Nodes are responsible for interpreting/executing their child nodes.
     nodes = [n for n in ws.nodes if not n.parentNode]
+    children = {}
+    for n in ws.nodes:
+        if n.parentNode:
+            children.setdefault(n.parentNode, []).append(n)
     outputs = {}
     failed = 0
     while len(outputs) + failed < len(nodes):
@@ -64,8 +68,14 @@ def execute(ws):
                 inputs = [outputs[input] for input in inputs]
                 data = node.data
                 op = ops.ALL_OPS[data.title]
+                params = {**data.params}
+                if op.sub_nodes:
+                    sub_nodes = children.get(node.id, [])
+                    sub_node_ids = [node.id for node in sub_nodes]
+                    sub_edges = [edge for edge in ws.edges if edge.source in sub_node_ids]
+                    params['sub_flow'] = {'nodes': sub_nodes, 'edges': sub_edges}
                 try:
-                  output = op(*inputs, **data.params)
+                  output = op(*inputs, **params)
                 except Exception as e:
                   traceback.print_exc()
                   data.error = str(e)

server/networkx_ops.py

@@ -22,9 +22,9 @@ def wrapped(func):
 
 
 for (name, func) in nx.__dict__.items():
-  if type(func) == nx.utils.backends._dispatch:
+  if hasattr(func, 'graphs'):
     sig = inspect.signature(func)
-    inputs = {'G': nx.Graph} if 'G' in sig.parameters else {}
+    inputs = {k: nx.Graph for k in func.graphs}
     params = {
       name:
         str(param.default)

server/ops.py

@@ -4,14 +4,33 @@ import functools
 import inspect
 import networkx as nx
 import pandas as pd
+import typing
 
 ALL_OPS = {}
+PARAM_TYPE = type[typing.Any]
+
+@dataclasses.dataclass
+class Parameter:
+  '''Defines a parameter for an operation.'''
+  name: str
+  default: any
+  type: PARAM_TYPE = None
+
+  def __post_init__(self):
+    if self.type is None:
+      self.type = type(self.default)
+  def to_json(self):
+    return {
+      'name': self.name,
+      'default': self.default,
+      'type': str(self.type),
+    }
 
 @dataclasses.dataclass
 class Op:
   func: callable
   name: str
-  params: dict # name -> default
+  params: dict[str, Parameter]
   inputs: dict # name -> type
   outputs: dict # name -> type
   type: str # The UI to use for this operation.
@@ -19,22 +38,17 @@ class Op:
 
   def __call__(self, *inputs, **params):
     # Convert parameters.
-    sig = inspect.signature(self.func)
-    for p in params:
+    for p in params.values():
       if p in self.params:
-        t = sig.parameters[p].annotation
-        if t is inspect._empty:
-          t = type(self.params[p])
-        if t == int:
+        if p.type == int:
           params[p] = int(params[p])
-        elif t == float:
+        elif p.type == float:
           params[p] = float(params[p])
     # Convert inputs.
     inputs = list(inputs)
-    for i, (x, p) in enumerate(zip(inputs, sig.parameters.values())):
-      t = p.annotation
+    for i, (x, t) in enumerate(zip(inputs, self.inputs.values())):
       if t == nx.Graph and isinstance(x, Bundle):
-        inputs[i] = o.to_nx()
+        inputs[i] = x.to_nx()
       elif t == Bundle and isinstance(x, nx.Graph):
         inputs[i] = Bundle.from_nx(x)
     res = self.func(*inputs, **params)
@@ -43,7 +57,7 @@ class Op:
   def to_json(self):
     return {
       'type': self.type,
-      'data': { 'title': self.name, 'params': self.params },
+      'data': { 'title': self.name, 'params': [p.to_json() for p in self.params.values()] },
       'targetPosition': 'left' if self.inputs else None,
       'sourcePosition': 'right' if self.outputs else None,
       'sub_nodes': [sub.to_json() for sub in self.sub_nodes.values()] if self.sub_nodes else None,
@@ -68,8 +82,8 @@ class Bundle:
   Can efficiently represent a knowledge graph (homogeneous or heterogeneous) or tabular data.
   It can also carry other data, such as a trained model.
   '''
-  dfs: dict
-  relations: list[RelationDefinition]
+  dfs: dict = dataclasses.field(default_factory=dict) # name -> DataFrame
+  relations: list[RelationDefinition] = dataclasses.field(default_factory=list)
   other: dict = None
 
   @classmethod
@@ -121,10 +135,15 @@ def op(name, *, view='basic', sub_nodes=None):
       name: param.annotation
       for name, param in sig.parameters.items()
       if param.kind != param.KEYWORD_ONLY}
-    params = {
-      name: param.default if param.default is not inspect._empty else None
-      for name, param in sig.parameters.items()
-      if param.kind == param.KEYWORD_ONLY}
+    params = {}
+    for n, param in sig.parameters.items():
+      if param.kind == param.KEYWORD_ONLY:
+        p = Parameter(n, param.default, param.annotation)
+        if p.default is inspect._empty:
+          p.default = None
+        if p.type is inspect._empty:
+          p.type = type(p.default)
+        params[n] = p
     outputs = {'output': 'yes'} if view == 'basic' else {} # Maybe more fancy later.
     op = Op(func, name, params=params, inputs=inputs, outputs=outputs, type=view)
     if sub_nodes is not None:
@@ -133,3 +152,13 @@ def op(name, *, view='basic', sub_nodes=None):
     ALL_OPS[name] = op
     return func
   return decorator
+
+def no_op(*args, **kwargs):
+  if args:
+    return args[0]
+  return Bundle()
+
+def register_passive_op(name, inputs={'input': Bundle}, outputs={'output': Bundle}, params=[]):
+  '''A passive operation has no associated code.'''
+  op = Op(no_op, name, params={p.name: p for p in params}, inputs=inputs, outputs=outputs, type='basic')
+  ALL_OPS[name] = op

server/pytorch_model_ops.py

@@ -1,4 +1,5 @@
 '''Boxes for defining and using PyTorch models.'''
+from enum import Enum
 import inspect
 from . import ops
 
@@ -6,9 +7,13 @@ LAYERS = {}
 
 @ops.op("Define PyTorch model", sub_nodes=LAYERS)
 def define_pytorch_model(*, sub_flow):
-  # import torch # Lazy import because it's slow.
   print('sub_flow:', sub_flow)
-  return 'hello ' + str(sub_flow)
+  return ops.Bundle(other={'model': str(sub_flow)})
+
+@ops.op("Train PyTorch model")
+def train_pytorch_model(model, graph):
+  # import torch # Lazy import because it's slow.
+  return 'hello ' + str(model)
 
 def register_layer(name):
   def decorator(func):
@@ -38,11 +43,22 @@ def dropout(*, p=0.5):
 def linear(*, output_dim: int):
   return f'Linear {output_dim}'
 
+class GraphConv(Enum):
+  GCNConv = 'GCNConv'
+  GATConv = 'GATConv'
+  GATv2Conv = 'GATv2Conv'
+  SAGEConv = 'SAGEConv'
+
 @register_layer('Graph Convolution')
-def graph_convolution():
+def graph_convolution(*, type: GraphConv):
   return 'GraphConv'
 
+class Nonlinearity(Enum):
+  Mish = 'Mish'
+  ReLU = 'ReLU'
+  Tanh = 'Tanh'
+
 @register_layer('Nonlinearity')
-def nonlinearity():
+def nonlinearity(*, type: Nonlinearity):
   return 'ReLU'
 

web/src/LynxKiteFlow.svelte

@@ -65,6 +65,8 @@
     nodes.update((n) => {
       node.position = screenToFlowPosition({x: nodeSearchSettings.pos.x, y: nodeSearchSettings.pos.y});
       const title = node.data.title;
+      node.data.params = Object.fromEntries(
+        node.data.params.map((p) => [p.name, p.default]));
       let i = 1;
       node.id = `${title} ${i}`;
       while (n.find((x) => x.id === node.id)) {

web/src/NodeWithTableView.svelte

@@ -27,6 +27,12 @@
       </table>
       {/if}
     {/each}
+    {#each Object.entries(data.view.others || {}) as [name, o]}
+      <div class="df-head" on:click={() => open[name] = !open[name]}>{name}</div>
+      {#if open[name]}
+      <pre>{o}</pre>
+      {/if}
+    {/each}
   {/if}
 </LynxKiteNode>
 <style>