Connect Nodes into a Pipeline
Ingest several nodes and wire them into a connected workflow graph using the returned node_ids and explicit origin_nodes / forward_nodes.
Instead of letting OptixLog auto-wire edges via a prompt, you can specify predecessor and successor nodes explicitly. This is useful when you know the exact shape of your workflow graph in code.
How explicit wiring works
The ingest overload that accepts origin_nodes and forward_nodes takes lists of node id strings:
project.ingest(
data=my_node,
origin_nodes=["node_id_a", "node_id_b"],
forward_nodes=["node_id_c"],
)origin_nodes— nodes that feed into this node (predecessors / parents).forward_nodes— nodes that this node feeds into (successors / children).
You obtain node ids from previous IngestResult.node_id values or from the Management API.
Building a three-node chain
This example builds a linear chain: LayoutNode → SimulationNode → MeasurementNode.
from optixlog_gen import (
OptixClient,
Projects,
SimulationNode,
MeasurementNode,
LayoutNode,
)
client = OptixClient()
grating = client.project(Projects.GRATING_COUPLER_LAB)
# Step 1: Ingest the layout (no predecessors yet).
layout_result = grating.ingest(
data=MeasurementNode(
instrument="VNA",
samples=[0.1, 0.2, 0.3],
notes="baseline measurement before simulation",
),
origin_nodes=[],
forward_nodes=[],
)
layout_id = layout_result.node_id
print(f"Layout node: {layout_id}")
# Step 2: Ingest the simulation, pointing back to the layout.
sim_result = grating.ingest(
data=SimulationNode(
config={"mesh": "fine"},
solver="fdtd",
wavelength_nm=1550.0,
),
origin_nodes=[layout_id],
forward_nodes=[],
)
sim_id = sim_result.node_id
print(f"Simulation node: {sim_id}")
# Step 3: Ingest a follow-up measurement, chained after the simulation.
meas_result = grating.ingest(
data=MeasurementNode(
instrument="OSA",
samples=[0.95, 0.96, 0.94],
notes="post-simulation optical spectrum",
),
origin_nodes=[sim_id],
forward_nodes=[],
)
meas_id = meas_result.node_id
print(f"Measurement node: {meas_id}")
client.close()After this runs, the workflow graph contains:
[MeasurementNode layout_id]
↓
[SimulationNode sim_id]
↓
[MeasurementNode meas_id]Wiring to an existing node
If you already know the id of a node from a previous run or from the Management API, pass it directly:
from optixlog.management import Management
mgmt = Management(client.sdk)
project_mgmt = mgmt.project("proj_grating_7f3a")
# Find the most recent FDTD simulation.
existing = project_mgmt.workflow_nodes.first(
has_property={"solver": "fdtd"},
)
existing_id = existing.id if existing else None
new_result = grating.ingest(
data=SimulationNode(config={"mesh": "coarse"}, solver="eme"),
origin_nodes=[existing_id] if existing_id else [],
forward_nodes=[],
)Branching graph
You can fan-out by supplying multiple forward_nodes, or fan-in by supplying multiple origin_nodes:
# Fan-out: this simulation feeds two downstream measurements.
sim_result = grating.ingest(
data=SimulationNode(config={}, solver="fdtd"),
origin_nodes=[],
forward_nodes=[meas_id_a, meas_id_b],
)prompt vs. explicit wiring
Use prompt when you want OptixLog to determine graph edges from the natural language description. Use origin_nodes / forward_nodes when your pipeline code knows the exact edges. The two modes are mutually exclusive — passing both raises a type error.
Reading back the wired edges
After ingesting, read the resolved edges from IngestResult:
print("Origins resolved:", [r.id for r in sim_result.origin_nodes])
print("Forward resolved:", [r.id for r in sim_result.forward_nodes])Or use the Management API to inspect the full node with parents and children:
node = project_mgmt.workflow_nodes.get(
sim_id,
include=["parents", "children", "properties"],
)
print("Parents:", [r.id for r in (node.parent_refs or [])])
print("Children:", [r.id for r in (node.child_refs or [])])