Merge pull request #85 from biggraph/darabos-nx

lynxkite-app/src/lynxkite_app/crdt.py

@@ -54,6 +54,10 @@ class WebsocketServer(pycrdt_websocket.WebsocketServer):
             # We have two possible sources of truth for the workspaces, the YStore and the JSON files.
             # In case we didn't find the workspace in the YStore, we try to load it from the JSON files.
             try_to_load_workspace(ws, name)
+        ws_simple = workspace.Workspace.model_validate(ws.to_py())
+        clean_input(ws_simple)
+        # Set the last known version to the current state, so we don't trigger a change event.
+        last_known_versions[name] = ws_simple
         room = pycrdt_websocket.YRoom(
             ystore=ystore, ydoc=ydoc, exception_handler=ws_exception_handler
         )
@@ -197,7 +201,6 @@ async def workspace_changed(name: str, changes: pycrdt.MapEvent, ws_crdt: pycrdt
         getattr(change, "keys", {}).get("__execution_delay", {}).get("newValue", 0)
         for change in changes
     )
-    print(f"Running {name} in {ws_pyd.env}...")
     if delay:
         task = asyncio.create_task(execute(name, ws_crdt, ws_pyd, delay))
         delayed_executions[name] = task
@@ -221,10 +224,12 @@ async def execute(
             await asyncio.sleep(delay)
         except asyncio.CancelledError:
             return
+    print(f"Running {name} in {ws_pyd.env}...")
     path = config.DATA_PATH / name
     assert path.is_relative_to(config.DATA_PATH), "Provided workspace path is invalid"
     # Save user changes before executing, in case the execution fails.
     workspace.save(ws_pyd, path)
+    ws_pyd._crdt = ws_crdt
     with ws_crdt.doc.transaction():
         for nc, np in zip(ws_crdt["nodes"], ws_pyd.nodes):
             if "data" not in nc:
@@ -234,6 +239,7 @@ async def execute(
             np._crdt = nc
     await workspace.execute(ws_pyd)
     workspace.save(ws_pyd, path)
+    print(f"Finished running {name} in {ws_pyd.env}.")
 
 
 @contextlib.asynccontextmanager

lynxkite-app/web/src/index.css

@@ -235,6 +235,25 @@ body {
       margin: 10px;
     }
   }
+
+  .env-select {
+    background: transparent;
+    color: #39bcf3;
+  }
+}
+
+.params-expander {
+  font-size: 15px;
+  padding: 4px;
+  color: #000a;
+}
+
+.flippy {
+  transition: transform 0.5s;
+}
+
+.flippy.flippy-90 {
+  transform: rotate(-90deg);
 }
 
 .directory {

lynxkite-app/web/src/workspace/EnvironmentSelector.tsx

@@ -6,7 +6,7 @@ export default function EnvironmentSelector(props: {
   return (
     <>
       <select
-        className="select w-full max-w-xs"
+        className="env-select select w-full max-w-xs"
         name="workspace-env"
         value={props.value}
         onChange={(evt) => props.onChange(evt.currentTarget.value)}

lynxkite-app/web/src/workspace/NodeSearch.tsx

@@ -25,9 +25,14 @@ export default function (props: {
       }),
     [props.boxes],
   );
+  const allOps = useMemo(() => {
+    const boxes = Object.values(props.boxes).map((box) => ({ item: box }));
+    boxes.sort((a, b) => a.item.name.localeCompare(b.item.name));
+    return boxes;
+  }, [props.boxes]);
   const hits: { item: OpsOp }[] = searchText
     ? fuse.search<OpsOp>(searchText)
-    : Object.values(props.boxes).map((box) => ({ item: box }));
+    : allOps;
   const [selectedIndex, setSelectedIndex] = useState(0);
   useEffect(() => searchBox.current.focus());
   function typed(text: string) {

lynxkite-app/web/src/workspace/Workspace.tsx

@@ -26,11 +26,11 @@ import { useParams } from "react-router";
 import useSWR, { type Fetcher } from "swr";
 import { WebsocketProvider } from "y-websocket";
 // @ts-ignore
-import ArrowBack from "~icons/tabler/arrow-back.jsx";
-// @ts-ignore
 import Atom from "~icons/tabler/atom.jsx";
 // @ts-ignore
 import Backspace from "~icons/tabler/backspace.jsx";
+// @ts-ignore
+import Close from "~icons/tabler/x.jsx";
 import type { Workspace, WorkspaceNode } from "../apiTypes.ts";
 import favicon from "../assets/favicon.ico";
 // import NodeWithTableView from './NodeWithTableView';
@@ -303,7 +303,7 @@ function LynxKiteFlow() {
             <Backspace />
           </a>
           <a href={`/dir/${parentDir}`}>
-            <ArrowBack />
+            <Close />
           </a>
         </div>
       </div>

lynxkite-app/web/src/workspace/nodes/NodeParameter.tsx

@@ -73,6 +73,10 @@ export default function NodeParameter({
             value={value || ""}
             onChange={(evt) => onChange(evt.currentTarget.value, { delay: 2 })}
             onBlur={(evt) => onChange(evt.currentTarget.value, { delay: 0 })}
+            onKeyDown={(evt) =>
+              evt.code === "Enter" &&
+              onChange(evt.currentTarget.value, { delay: 0 })
+            }
           />
         </>
       )}

lynxkite-app/web/src/workspace/nodes/NodeWithParams.tsx

@@ -1,4 +1,7 @@
 import { useReactFlow } from "@xyflow/react";
+import React from "react";
+// @ts-ignore
+import Triangle from "~icons/tabler/triangle-inverted-filled.jsx";
 import LynxKiteNode from "./LynxKiteNode";
 import NodeGroupParameter from "./NodeGroupParameter";
 import NodeParameter from "./NodeParameter";
@@ -8,6 +11,7 @@ export type UpdateOptions = { delay?: number };
 function NodeWithParams(props: any) {
   const reactFlow = useReactFlow();
   const metaParams = props.data.meta?.params;
+  const [collapsed, setCollapsed] = React.useState(props.collapsed);
 
   function setParam(name: string, newValue: any, opts: UpdateOptions) {
     reactFlow.updateNodeData(props.id, (prevData: any) => ({
@@ -31,31 +35,40 @@ function NodeWithParams(props: any) {
 
   return (
     <LynxKiteNode {...props}>
-      {params.map(([name, value]) =>
-        metaParams?.[name]?.type === "group" ? (
-          <NodeGroupParameter
-            key={name}
-            value={value}
-            meta={metaParams?.[name]}
-            setParam={(name: string, value: any, opts?: UpdateOptions) =>
-              setParam(name, value, opts || {})
-            }
-            deleteParam={(name: string, opts?: UpdateOptions) =>
-              deleteParam(name, opts || {})
-            }
-          />
-        ) : (
-          <NodeParameter
-            name={name}
-            key={name}
-            value={value}
-            meta={metaParams?.[name]}
-            onChange={(value: any, opts?: UpdateOptions) =>
-              setParam(name, value, opts || {})
-            }
-          />
-        ),
+      {props.collapsed && (
+        <div
+          className="params-expander"
+          onClick={() => setCollapsed(!collapsed)}
+        >
+          <Triangle className={`flippy ${collapsed ? "flippy-90" : ""}`} />
+        </div>
       )}
+      {!collapsed &&
+        params.map(([name, value]) =>
+          metaParams?.[name]?.type === "group" ? (
+            <NodeGroupParameter
+              key={name}
+              value={value}
+              meta={metaParams?.[name]}
+              setParam={(name: string, value: any, opts?: UpdateOptions) =>
+                setParam(name, value, opts || {})
+              }
+              deleteParam={(name: string, opts?: UpdateOptions) =>
+                deleteParam(name, opts || {})
+              }
+            />
+          ) : (
+            <NodeParameter
+              name={name}
+              key={name}
+              value={value}
+              meta={metaParams?.[name]}
+              onChange={(value: any, opts?: UpdateOptions) =>
+                setParam(name, value, opts || {})
+              }
+            />
+          ),
+        )}
       {props.children}
     </LynxKiteNode>
   );

lynxkite-app/web/src/workspace/nodes/NodeWithVisualization.tsx

@@ -10,20 +10,28 @@ const NodeWithVisualization = (props: any) => {
     if (!opts || !chartsRef.current) return;
     chartsInstanceRef.current = echarts.init(chartsRef.current, null, {
       renderer: "canvas",
-      width: 800,
-      height: 800,
+      width: "auto",
+      height: "auto",
     });
     chartsInstanceRef.current.setOption(opts);
-    const onResize = () => chartsInstanceRef.current?.resize();
-    window.addEventListener("resize", onResize);
+    const resizeObserver = new ResizeObserver(() => {
+      const e = chartsRef.current!;
+      e.style.padding = "1px";
+      chartsInstanceRef.current?.resize();
+      e.style.padding = "0";
+    });
+    const observed = chartsRef.current;
+    resizeObserver.observe(observed);
     return () => {
-      window.removeEventListener("resize", onResize);
+      resizeObserver.unobserve(observed);
       chartsInstanceRef.current?.dispose();
     };
   }, [props.data?.display?.value]);
+  const nodeStyle = { display: "flex", flexDirection: "column" };
+  const vizStyle = { flex: 1 };
   return (
-    <NodeWithParams {...props}>
-      <div className="box" draggable={false} ref={chartsRef} />
+    <NodeWithParams nodeStyle={nodeStyle} collapsed {...props}>
+      <div style={vizStyle} ref={chartsRef} />
     </NodeWithParams>
   );
 };

lynxkite-app/web/tests/basic.spec.ts

@@ -35,9 +35,9 @@ test("Box creation & deletion per env", async () => {
 });
 
 test("Delete multi-handle boxes", async () => {
-  await workspace.addBox("Compute PageRank");
-  await workspace.deleteBoxes(["Compute PageRank 1"]);
-  await expect(workspace.getBox("Compute PageRank 1")).not.toBeVisible();
+  await workspace.addBox("NX › PageRank");
+  await workspace.deleteBoxes(["NX › PageRank 1"]);
+  await expect(workspace.getBox("NX › PageRank 1")).not.toBeVisible();
 });
 
 test("Drag box", async () => {

lynxkite-app/web/tests/errors.spec.ts

@@ -20,24 +20,24 @@ test.afterEach(async () => {
 test("missing parameter", async () => {
   // Test the correct error message is displayed when a required parameter is missing,
   // and that the error message is removed when the parameter is filled.
-  await workspace.addBox("Create scale-free graph");
-  const graphBox = workspace.getBox("Create scale-free graph 1");
-  await graphBox.locator("input").fill("");
-  expect(await graphBox.locator(".error").innerText()).toBe(
-    "invalid literal for int() with base 10: ''",
-  );
-  await graphBox.locator("input").fill("10");
+  await workspace.addBox("NX › Scale-Free Graph");
+  const graphBox = workspace.getBox("NX › Scale-Free Graph 1");
+  await expect(graphBox.locator(".error")).toHaveText("n is unset.");
+  await graphBox.getByLabel("n", { exact: true }).fill("10");
   await expect(graphBox.locator(".error")).not.toBeVisible();
 });
 
 test("unknown operation", async () => {
   // Test that the correct error is displayed when the operation does not belong to
   // the current environment.
-  await workspace.addBox("Create scale-free graph");
+  await workspace.addBox("NX › Scale-Free Graph");
+  const graphBox = workspace.getBox("NX › Scale-Free Graph 1");
+  await graphBox.getByLabel("n", { exact: true }).fill("10");
   await workspace.setEnv("LynxScribe");
-  const csvBox = workspace.getBox("Create scale-free graph 1");
-  const errorText = await csvBox.locator(".error").innerText();
-  expect(errorText).toBe('Operation "Create scale-free graph" not found.');
+  const csvBox = workspace.getBox("NX › Scale-Free Graph 1");
+  await expect(csvBox.locator(".error")).toHaveText(
+    'Operation "NX › Scale-Free Graph" not found.',
+  );
   await workspace.setEnv("LynxKite Graph Analytics");
   await expect(csvBox.locator(".error")).not.toBeVisible();
 });

lynxkite-app/web/tests/examples.spec.ts

@@ -23,13 +23,13 @@ test.fail("AIMO example", async ({ page }) => {
   await ws.expectErrorFree();
 });
 
-test.fail("LynxScribe example", async ({ page }) => {
+test("LynxScribe example", async ({ page }) => {
   // Fails because of missing OPENAI_API_KEY
   const ws = await Workspace.open(page, "LynxScribe demo");
   await ws.expectErrorFree();
 });
 
-test.fail("Graph RAG", async ({ page }) => {
+test("Graph RAG", async ({ page }) => {
   // Fails due to some issue with ChromaDB
   const ws = await Workspace.open(page, "Graph RAG");
   await ws.expectErrorFree(process.env.CI ? 2000 : 500);

lynxkite-app/web/tests/graph_creation.spec.ts

@@ -9,9 +9,13 @@ test.beforeEach(async ({ browser }) => {
     await browser.newPage(),
     "graph_creation_spec_test",
   );
-  await workspace.addBox("Create scale-free graph");
+  await workspace.addBox("NX › Scale-Free Graph");
+  await workspace
+    .getBox("NX › Scale-Free Graph 1")
+    .getByLabel("n", { exact: true })
+    .fill("10");
   await workspace.addBox("Create graph");
-  await workspace.connectBoxes("Create scale-free graph 1", "Create graph 1");
+  await workspace.connectBoxes("NX › Scale-Free Graph 1", "Create graph 1");
 });
 
 test.afterEach(async () => {

lynxkite-app/web/tests/lynxkite.ts

@@ -65,7 +65,10 @@ export class Workspace {
     // Some x,y offset, otherwise the box handle may fall outside the viewport.
     await this.page.locator(".ws-name").click();
     await this.page.keyboard.press("/");
-    await this.page.locator(".node-search").getByText(boxName).click();
+    await this.page
+      .locator(".node-search")
+      .getByText(boxName, { exact: true })
+      .click();
     await expect(this.getBoxes()).toHaveCount(allBoxes.length + 1);
   }
 

lynxkite-core/src/lynxkite/core/executors/one_by_one.py

@@ -142,12 +142,12 @@ async def execute(ws: workspace.Workspace, catalog, cache=None):
                         key = make_cache_key((inputs, params))
                         if key not in cache:
                             result: ops.Result = op(*inputs, **params)
-                            output = await await_if_needed(result.output)
-                            cache[key] = output
-                        output = cache[key]
+                            result.output = await await_if_needed(result.output)
+                            cache[key] = result
+                        result = cache[key]
                     else:
                         result = op(*inputs, **params)
-                        output = await await_if_needed(result.output)
+                    output = await await_if_needed(result.output)
                 except Exception as e:
                     traceback.print_exc()
                     node.publish_error(e)

lynxkite-core/src/lynxkite/core/ops.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 import enum
 import functools
 import inspect
+import types
 import pydantic
 import typing
 from dataclasses import dataclass
@@ -123,6 +124,25 @@ def basic_outputs(*names):
     return {name: Output(name=name, type=None) for name in names}
 
 
+def _param_to_type(name, value, type):
+    value = value or ""
+    if type is int:
+        assert value != "", f"{name} is unset."
+        return int(value)
+    if type is float:
+        assert value != "", f"{name} is unset."
+        return float(value)
+    if isinstance(type, enum.EnumMeta):
+        return type[value]
+    if isinstance(type, types.UnionType):
+        match type.__args__:
+            case (types.NoneType, type):
+                return None if value == "" else _param_to_type(name, value, type)
+            case (type, types.NoneType):
+                return None if value == "" else _param_to_type(name, value, type)
+    return value
+
+
 class Op(BaseConfig):
     func: typing.Callable = pydantic.Field(exclude=True)
     name: str
@@ -136,12 +156,7 @@ class Op(BaseConfig):
         # Convert parameters.
         for p in params:
             if p in self.params:
-                if self.params[p].type is int:
-                    params[p] = int(params[p])
-                elif self.params[p].type is float:
-                    params[p] = float(params[p])
-                elif isinstance(self.params[p].type, enum.EnumMeta):
-                    params[p] = self.params[p].type[params[p]]
+                params[p] = _param_to_type(p, params[p], self.params[p].type)
         res = self.func(*inputs, **params)
         if not isinstance(res, Result):
             # Automatically wrap the result in a Result object, if it isn't already.

lynxkite-core/src/lynxkite/core/workspace.py

@@ -92,6 +92,7 @@ class Workspace(BaseConfig):
     env: str = ""
     nodes: list[WorkspaceNode] = dataclasses.field(default_factory=list)
     edges: list[WorkspaceEdge] = dataclasses.field(default_factory=list)
+    _crdt: pycrdt.Map
 
 
 async def execute(ws: Workspace):

lynxkite-graph-analytics/src/lynxkite_graph_analytics/core.py

@@ -1,6 +1,7 @@
 """Graph analytics executor and data types."""
 
-from lynxkite.core import ops
+import os
+from lynxkite.core import ops, workspace
 import dataclasses
 import functools
 import networkx as nx
@@ -134,54 +135,73 @@ def nx_node_attribute_func(name):
     return decorator
 
 
-def disambiguate_edges(ws):
+def disambiguate_edges(ws: workspace.Workspace):
     """If an input plug is connected to multiple edges, keep only the last edge."""
     seen = set()
     for edge in reversed(ws.edges):
         if (edge.target, edge.targetHandle) in seen:
-            ws.edges.remove(edge)
+            i = ws.edges.index(edge)
+            del ws.edges[i]
+            if hasattr(ws, "_crdt"):
+                del ws._crdt["edges"][i]
         seen.add((edge.target, edge.targetHandle))
 
 
 @ops.register_executor(ENV)
-async def execute(ws):
+async def execute(ws: workspace.Workspace):
     catalog: dict[str, ops.Op] = ops.CATALOGS[ws.env]
     disambiguate_edges(ws)
     outputs = {}
-    failed = 0
-    while len(outputs) + failed < len(ws.nodes):
-        for node in ws.nodes:
-            if node.id in outputs:
-                continue
-            # TODO: Take the input/output handles into account.
-            inputs = [edge.source for edge in ws.edges if edge.target == node.id]
-            if all(input in outputs for input in inputs):
+    nodes = {node.id: node for node in ws.nodes}
+    todo = set(nodes.keys())
+    progress = True
+    while progress:
+        progress = False
+        for id in list(todo):
+            node = nodes[id]
+            input_nodes = [edge.source for edge in ws.edges if edge.target == id]
+            if all(input in outputs for input in input_nodes):
                 # All inputs for this node are ready, we can compute the output.
-                inputs = [outputs[input] for input in inputs]
-                params = {**node.data.params}
-                op = catalog.get(node.data.title)
-                if not op:
-                    node.publish_error("Operation not found in catalog")
-                    failed += 1
-                    continue
-                node.publish_started()
-                try:
-                    # Convert inputs types  to match operation signature.
-                    for i, (x, p) in enumerate(zip(inputs, op.inputs.values())):
-                        if p.type == nx.Graph and isinstance(x, Bundle):
-                            inputs[i] = x.to_nx()
-                        elif p.type == Bundle and isinstance(x, nx.Graph):
-                            inputs[i] = Bundle.from_nx(x)
-                        elif p.type == Bundle and isinstance(x, pd.DataFrame):
-                            inputs[i] = Bundle.from_df(x)
-                    result = op(*inputs, **params)
-                except Exception as e:
-                    traceback.print_exc()
-                    node.publish_error(e)
-                    failed += 1
-                    continue
-                outputs[node.id] = result.output
-                node.publish_result(result)
+                todo.remove(id)
+                progress = True
+                _execute_node(node, ws, catalog, outputs)
+
+
+def _execute_node(node, ws, catalog, outputs):
+    params = {**node.data.params}
+    op = catalog.get(node.data.title)
+    if not op:
+        node.publish_error("Operation not found in catalog")
+        return
+    node.publish_started()
+    input_map = {
+        edge.targetHandle: outputs[edge.source]
+        for edge in ws.edges
+        if edge.target == node.id
+    }
+    try:
+        # Convert inputs types to match operation signature.
+        inputs = []
+        for p in op.inputs.values():
+            if p.name not in input_map:
+                node.publish_error(f"Missing input: {p.name}")
+                return
+            x = input_map[p.name]
+            if p.type == nx.Graph and isinstance(x, Bundle):
+                x = x.to_nx()
+            elif p.type == Bundle and isinstance(x, nx.Graph):
+                x = Bundle.from_nx(x)
+            elif p.type == Bundle and isinstance(x, pd.DataFrame):
+                x = Bundle.from_df(x)
+            inputs.append(x)
+        result = op(*inputs, **params)
+    except Exception as e:
+        if os.environ.get("LYNXKITE_LOG_OP_ERRORS"):
+            traceback.print_exc()
+        node.publish_error(e)
+        return
+    outputs[node.id] = result.output
+    node.publish_result(result)
 
 
 def df_for_frontend(df: pd.DataFrame, limit: int) -> pd.DataFrame:

lynxkite-graph-analytics/src/lynxkite_graph_analytics/lynxkite_ops.py

@@ -122,25 +122,6 @@ def import_osm(*, location: str):
     return ox.graph.graph_from_place(location, network_type="drive")
 
 
-@op("Create scale-free graph")
-def create_scale_free_graph(*, nodes: int = 10):
-    """Creates a scale-free graph with the given number of nodes."""
-    return nx.scale_free_graph(nodes)
-
-
-@op("Compute PageRank")
-@core.nx_node_attribute_func("pagerank")
-def compute_pagerank(graph: nx.Graph, *, damping=0.85, iterations=100):
-    # TODO: This requires scipy to be installed.
-    return nx.pagerank(graph, alpha=damping, max_iter=iterations)
-
-
-@op("Compute betweenness centrality")
-@core.nx_node_attribute_func("betweenness_centrality")
-def compute_betweenness_centrality(graph: nx.Graph, *, k=10):
-    return nx.betweenness_centrality(graph, k=k)
-
-
 @op("Discard loop edges")
 def discard_loop_edges(graph: nx.Graph):
     graph = graph.copy()

lynxkite-graph-analytics/src/lynxkite_graph_analytics/networkx_ops.py

@@ -1,13 +1,155 @@
 """Automatically wraps all NetworkX functions as LynxKite operations."""
 
+import collections
+import types
 from lynxkite.core import ops
 import functools
 import inspect
 import networkx as nx
+import re
+
+import pandas as pd
 
 ENV = "LynxKite Graph Analytics"
 
 
+class UnsupportedParameterType(Exception):
+    pass
+
+
+_UNSUPPORTED = object()
+_SKIP = object()
+
+
+def doc_to_type(name: str, type_hint: str) -> type:
+    type_hint = type_hint.lower()
+    type_hint = re.sub("[(][^)]+[)]", "", type_hint).strip().strip(".")
+    if " " in name or "http" in name:
+        return _UNSUPPORTED  # Not a parameter type.
+    if type_hint.endswith(", optional"):
+        w = doc_to_type(name, type_hint.removesuffix(", optional").strip())
+        if w is _UNSUPPORTED:
+            return _SKIP
+        return w if w is _SKIP else w | None
+    if type_hint in [
+        "a digraph or multidigraph",
+        "a graph g",
+        "graph",
+        "graphs",
+        "networkx graph instance",
+        "networkx graph",
+        "networkx undirected graph",
+        "nx.graph",
+        "undirected graph",
+        "undirected networkx graph",
+    ] or type_hint.startswith("networkx graph"):
+        return nx.Graph
+    elif type_hint in [
+        "digraph-like",
+        "digraph",
+        "directed graph",
+        "networkx digraph",
+        "networkx directed graph",
+        "nx.digraph",
+    ]:
+        return nx.DiGraph
+    elif type_hint == "node":
+        return _UNSUPPORTED
+    elif type_hint == '"node (optional)"':
+        return _SKIP
+    elif type_hint == '"edge"':
+        return _UNSUPPORTED
+    elif type_hint == '"edge (optional)"':
+        return _SKIP
+    elif type_hint in ["class", "data type"]:
+        return _UNSUPPORTED
+    elif type_hint in ["string", "str", "node label"]:
+        return str
+    elif type_hint in ["string or none", "none or string", "string, or none"]:
+        return str | None
+    elif type_hint in ["int", "integer"]:
+        return int
+    elif type_hint in ["bool", "boolean"]:
+        return bool
+    elif type_hint == "tuple":
+        return _UNSUPPORTED
+    elif type_hint == "set":
+        return _UNSUPPORTED
+    elif type_hint == "list of floats":
+        return _UNSUPPORTED
+    elif type_hint == "list of floats or float":
+        return float
+    elif type_hint in ["dict", "dictionary"]:
+        return _UNSUPPORTED
+    elif type_hint == "scalar or dictionary":
+        return float
+    elif type_hint == "none or dict":
+        return _SKIP
+    elif type_hint in ["function", "callable"]:
+        return _UNSUPPORTED
+    elif type_hint in [
+        "collection",
+        "container of nodes",
+        "list of nodes",
+    ]:
+        return _UNSUPPORTED
+    elif type_hint in [
+        "container",
+        "generator",
+        "iterable",
+        "iterator",
+        "list or iterable container",
+        "list or iterable",
+        "list or set",
+        "list or tuple",
+        "list",
+    ]:
+        return _UNSUPPORTED
+    elif type_hint == "generator of sets":
+        return _UNSUPPORTED
+    elif type_hint == "dict or a set of 2 or 3 tuples":
+        return _UNSUPPORTED
+    elif type_hint == "set of 2 or 3 tuples":
+        return _UNSUPPORTED
+    elif type_hint == "none, string or function":
+        return str | None
+    elif type_hint == "string or function" and name == "weight":
+        return str
+    elif type_hint == "integer, float, or none":
+        return float | None
+    elif type_hint in [
+        "float",
+        "int or float",
+        "integer or float",
+        "integer, float",
+        "number",
+        "numeric",
+        "real",
+        "scalar",
+    ]:
+        return float
+    elif type_hint in ["integer or none", "int or none"]:
+        return int | None
+    elif name == "seed":
+        return int | None
+    elif name == "weight":
+        return str
+    elif type_hint == "object":
+        return _UNSUPPORTED
+    return _SKIP
+
+
+def types_from_doc(doc: str) -> dict[str, type]:
+    types = {}
+    for line in doc.splitlines():
+        if ":" in line:
+            a, b = line.split(":", 1)
+            for a in a.split(","):
+                a = a.strip()
+                types[a] = doc_to_type(a, b)
+    return types
+
+
 def wrapped(name: str, func):
     @functools.wraps(func)
     def wrapper(*args, **kwargs):
@@ -15,48 +157,118 @@ def wrapped(name: str, func):
             if v == "None":
                 kwargs[k] = None
         res = func(*args, **kwargs)
+        # Figure out what the returned value is.
         if isinstance(res, nx.Graph):
             return res
-        # Otherwise it's a node attribute.
-        graph = args[0].copy()
-        nx.set_node_attributes(graph, values=res, name=name)
-        return graph
+        if isinstance(res, types.GeneratorType):
+            res = list(res)
+        if name in ["articulation_points"]:
+            graph = args[0].copy()
+            nx.set_node_attributes(graph, 0, name=name)
+            nx.set_node_attributes(graph, {r: 1 for r in res}, name=name)
+            return graph
+        if isinstance(res, collections.abc.Sized):
+            if len(res) == 0:
+                return pd.DataFrame()
+            for a in args:
+                if isinstance(a, nx.Graph):
+                    if a.number_of_nodes() == len(res):
+                        graph = a.copy()
+                        nx.set_node_attributes(graph, values=res, name=name)
+                        return graph
+                    if a.number_of_edges() == len(res):
+                        graph = a.copy()
+                        nx.set_edge_attributes(graph, values=res, name=name)
+                        return graph
+            return pd.DataFrame({name: res})
+        return pd.DataFrame({name: [res]})
 
     return wrapper
 
 
+def _get_params(func) -> dict | None:
+    sig = inspect.signature(func)
+    # Get types from docstring.
+    types = types_from_doc(func.__doc__)
+    # Always hide these.
+    for k in ["backend", "backend_kwargs", "create_using"]:
+        types[k] = _SKIP
+    # Add in types based on signature.
+    for k, param in sig.parameters.items():
+        if k in types:
+            continue
+        if param.annotation is not param.empty:
+            types[k] = param.annotation
+        if k in ["i", "j", "n"]:
+            types[k] = int
+    params = {}
+    for name, param in sig.parameters.items():
+        _type = types.get(name, _UNSUPPORTED)
+        if _type is _UNSUPPORTED:
+            raise UnsupportedParameterType(name)
+        if _type is _SKIP or _type in [nx.Graph, nx.DiGraph]:
+            continue
+        params[name] = ops.Parameter.basic(
+            name=name,
+            default=str(param.default)
+            if type(param.default) in [str, int, float]
+            else None,
+            type=_type,
+        )
+    return params
+
+
+_REPLACEMENTS = [
+    ("Barabasi Albert", "Barabasi–Albert"),
+    ("Bellman Ford", "Bellman–Ford"),
+    ("Bethe Hessian", "Bethe–Hessian"),
+    ("Bfs", "BFS"),
+    ("Dag ", "DAG "),
+    ("Dfs", "DFS"),
+    ("Dorogovtsev Goltsev Mendes", "Dorogovtsev–Goltsev–Mendes"),
+    ("Erdos Renyi", "Erdos–Renyi"),
+    ("Floyd Warshall", "Floyd–Warshall"),
+    ("Gnc", "G(n,c)"),
+    ("Gnm", "G(n,m)"),
+    ("Gnp", "G(n,p)"),
+    ("Gnr", "G(n,r)"),
+    ("Havel Hakimi", "Havel–Hakimi"),
+    ("Hkn", "H(k,n)"),
+    ("Hnm", "H(n,m)"),
+    ("Kl ", "KL "),
+    ("Moebius Kantor", "Moebius–Kantor"),
+    ("Pagerank", "PageRank"),
+    ("Scale Free", "Scale-Free"),
+    ("Vf2Pp", "VF2++"),
+    ("Watts Strogatz", "Watts–Strogatz"),
+    ("Weisfeiler Lehman", "Weisfeiler–Lehman"),
+]
+
+
 def register_networkx(env: str):
     cat = ops.CATALOGS.setdefault(env, {})
+    counter = 0
     for name, func in nx.__dict__.items():
         if hasattr(func, "graphs"):
-            sig = inspect.signature(func)
+            try:
+                params = _get_params(func)
+            except UnsupportedParameterType:
+                continue
             inputs = {k: ops.Input(name=k, type=nx.Graph) for k in func.graphs}
-            params = {
-                name: ops.Parameter.basic(
-                    name,
-                    str(param.default)
-                    if type(param.default) in [str, int, float]
-                    else None,
-                    param.annotation,
-                )
-                for name, param in sig.parameters.items()
-                if name not in ["G", "backend", "backend_kwargs", "create_using"]
-            }
-            for p in params.values():
-                if not p.type:
-                    # Guess the type based on the name.
-                    if len(p.name) == 1:
-                        p.type = int
-            name = "NX › " + name.replace("_", " ").title()
+            nicename = "NX › " + name.replace("_", " ").title()
+            for a, b in _REPLACEMENTS:
+                nicename = nicename.replace(a, b)
             op = ops.Op(
                 func=wrapped(name, func),
-                name=name,
+                name=nicename,
                 params=params,
                 inputs=inputs,
                 outputs={"output": ops.Output(name="output", type=nx.Graph)},
                 type="basic",
             )
-            cat[name] = op
+            cat[nicename] = op
+            counter += 1
+    print(f"Registered {counter} NetworkX operations.")
 
 
 register_networkx(ENV)

lynxkite-graph-analytics/tests/test_lynxkite_ops.py

@@ -77,13 +77,13 @@ async def test_execute_operation_inputs_correct_cast():
     )
     ws.edges = [
         workspace.WorkspaceEdge(
-            id="1", source="1", target="2", sourceHandle="1", targetHandle="2"
+            id="1", source="1", target="2", sourceHandle="output", targetHandle="graph"
         ),
         workspace.WorkspaceEdge(
-            id="2", source="2", target="3", sourceHandle="2", targetHandle="3"
+            id="2", source="2", target="3", sourceHandle="output", targetHandle="bundle"
         ),
         workspace.WorkspaceEdge(
-            id="3", source="3", target="4", sourceHandle="3", targetHandle="4"
+            id="3", source="3", target="4", sourceHandle="output", targetHandle="bundle"
         ),
     ]
 
@@ -92,5 +92,73 @@ async def test_execute_operation_inputs_correct_cast():
     assert all([node.data.error is None for node in ws.nodes])
 
 
+async def test_multiple_inputs():
+    """Make sure each input goes to the right argument."""
+    op = ops.op_registration("test")
+
+    @op("One")
+    def one():
+        return 1
+
+    @op("Two")
+    def two():
+        return 2
+
+    @op("Smaller?", view="visualization")
+    def is_smaller(a, b):
+        return a < b
+
+    ws = workspace.Workspace(env="test")
+    ws.nodes.append(
+        workspace.WorkspaceNode(
+            id="one",
+            type="cool",
+            data=workspace.WorkspaceNodeData(title="One", params={}),
+            position=workspace.Position(x=0, y=0),
+        )
+    )
+    ws.nodes.append(
+        workspace.WorkspaceNode(
+            id="two",
+            type="cool",
+            data=workspace.WorkspaceNodeData(title="Two", params={}),
+            position=workspace.Position(x=100, y=0),
+        )
+    )
+    ws.nodes.append(
+        workspace.WorkspaceNode(
+            id="smaller",
+            type="cool",
+            data=workspace.WorkspaceNodeData(title="Smaller?", params={}),
+            position=workspace.Position(x=200, y=0),
+        )
+    )
+    ws.edges = [
+        workspace.WorkspaceEdge(
+            id="one",
+            source="one",
+            target="smaller",
+            sourceHandle="output",
+            targetHandle="a",
+        ),
+        workspace.WorkspaceEdge(
+            id="two",
+            source="two",
+            target="smaller",
+            sourceHandle="output",
+            targetHandle="b",
+        ),
+    ]
+
+    await execute(ws)
+
+    assert ws.nodes[-1].data.display is True
+    # Flip the inputs.
+    ws.edges[0].targetHandle = "b"
+    ws.edges[1].targetHandle = "a"
+    await execute(ws)
+    assert ws.nodes[-1].data.display is False
+
+
 if __name__ == "__main__":
     pytest.main()

lynxkite-lynxscribe/README.md

@@ -5,7 +5,7 @@ LynxKite UI for building LynxScribe chat applications. Also runs the chat applic
 To run a chat UI for LynxScribe workspaces:
 
 ```bash
-WEBUI_AUTH=false OPENAI_API_BASE_URL=http://localhost:8000/api/service/lynxscribe/lynxscribe_ops uvx open-webui serve
+WEBUI_AUTH=false OPENAI_API_BASE_URL=http://localhost:8000/api/service/lynxkite_lynxscribe uvx open-webui serve
 ```
 
 Or use [Lynx WebUI](https://github.com/biggraph/lynx-webui/) instead of Open WebUI.

lynxkite-lynxscribe/src/lynxkite_lynxscribe/lynxscribe_ops.py

@@ -2,6 +2,8 @@
 LynxScribe configuration and testing in LynxKite.
 """
 
+import os
+import pathlib
 from lynxscribe.core.llm.base import get_llm_engine
 from lynxscribe.core.vector_store.base import get_vector_store
 from lynxscribe.common.config import load_config
@@ -221,10 +223,8 @@ def view(input):
 
 
 async def get_chat_api(ws):
-    import pathlib
     from lynxkite.core import workspace
 
-    DATA_PATH = pathlib.Path.cwd() / "data"
     path = DATA_PATH / ws
     assert path.is_relative_to(DATA_PATH)
     assert path.exists(), f"Workspace {path} does not exist"
@@ -285,18 +285,19 @@ async def api_service_get(request):
     return {"error": "Not found"}
 
 
+DATA_PATH = pathlib.Path(os.environ.get("LYNXKITE_DATA", "lynxkite_data"))
+
+
 def get_lynxscribe_workspaces():
-    import pathlib
     from lynxkite.core import workspace
 
-    DATA_DIR = pathlib.Path.cwd() / "data"
     workspaces = []
-    for p in DATA_DIR.glob("**/*"):
+    for p in DATA_PATH.glob("**/*"):
         if p.is_file():
             try:
                 ws = workspace.load(p)
                 if ws.env == ENV:
-                    workspaces.append(p.relative_to(DATA_DIR))
+                    workspaces.append(p.relative_to(DATA_PATH))
             except Exception:
                 pass  # Ignore files that are not valid workspaces.
     workspaces.sort()