feat: Process parameters (#174)

toby-coleman · web-flow · commit 1b5b079421ab · 2025-12-03T20:32:15.000Z
# Summary Closes #97 by implementing process parameters, which are shared via the `StateBackend` with the components. This allows users to reuse configuration across many components without duplication in the config. # Changes * Process parameters; * Updates to component to retrieve parameters once connected to the backend; * Update to `Tuner` to allow optimisation over process parameters; * Updated example in documentation.
diff --git a/docs/examples/tutorials/more-complex-process.md b/docs/examples/tutorials/more-complex-process.md
@@ -83,6 +83,24 @@ We'll provide an initial input value of `a = 0` to the `Scale` component, allowi
     
     Setting `initial_values = {"a": [0]}` means that we want the `a` input to be set to `0` on the first step and then revert to reading its input as usual.
 
+## Using parameters
+
+Sometimes you might need to use a parameter value repeatedly across many [`Component`][plugboard.component.Component] objects. The [`Process`][plugboard.process.Process] object allows you to supply a single dictionary of parameter values that are then supplied to all the components before they start running.
+For example, we can rewrite the `Scale` component like this:
+
+```python
+--8<-- "examples/tutorials/002_complex_processes/components.py:parameters"
+```
+
+Now we can supply the `scale` value when we create the [`Process`][plugboard.process.Process], and use it many places within the model.
+
+```python
+--8<-- "examples/tutorials/002_complex_processes/parameters.py:main"
+```
+
+1.  Supply a dictionary of common parameter values to the `Process` here. They are accessible from within all components.
+2.  If you need to override a parameter on a specific component, you can supply it via the component-level `parameters` argument.
+
 ## Next steps
 
 You've now learned how to build up complex model layouts in Plugboard. In the next tutorial we'll show how powerful a Plugboard model can be as we start to include different types of [`Component`][plugboard.component.Component].
diff --git a/docs/examples/tutorials/tuning-a-process.md b/docs/examples/tutorials/tuning-a-process.md
@@ -61,6 +61,9 @@ Running this code will execute an optimisation job and print out information on
 !!! tip
     You can impose arbitary constraints on variables within a `Process`. In your `step` method you can raise a [`ConstraintError`][plugboard.exceptions.ConstraintError] to indicate to the `Tuner` that a constraint has been breached. This will cause the trial to be stopped, and the optimisation will continue trying to find parameters that don't cause the constraint violation.
 
+!!! tip
+    You can optimise over process parameters if you have them in your model. Set `object_type="process"` and `field_type="parameter"` when specifying your tunable parameter.
+
 ## Using YAML config
 
 Plugboard's YAML config supports an optional `tune` section, allowing you to define optimisation jobs alongside your model configuration:
diff --git a/examples/demos/fundamentals/002_production_line_optimisation/production-line.ipynb b/examples/demos/fundamentals/002_production_line_optimisation/production-line.ipynb
@@ -347,11 +347,10 @@
     "\n",
     "```\n",
     "Best parameters found:\n",
-    "Config: {'controller1.threshold': 10, 'controller2.threshold': 38} - Metrics: {'cost_per_unit.cost_per_unit': 22.491974317817014, 'timestamp': 1755022664, \n",
-    "'checkpoint_dir_name': None, 'done': True, 'training_iteration': 1, 'trial_id': '287aff0b', 'date': '2025-08-12_19-17-44', 'time_this_iter_s': 2.224583864212036, \n",
-    "'time_total_s': 2.224583864212036, 'pid': 94765, 'hostname': 'hostname.local', 'node_ip': '127.0.0.1', 'config': {'controller1.threshold': 10, \n",
-    "'controller2.threshold': 38}, 'time_since_restore': 2.224583864212036, 'iterations_since_restore': 1, 'experiment_tag': \n",
-    "'14_controller1_threshold=10,controller2_threshold=38'}\n",
+    "Config: {'component.controller1.arg.threshold': 10, 'component.controller2.arg.threshold': 38} - Metrics: {'component.cost_per_unit.field.cost_per_unit': 22.491974317817014, 'timestamp': 1763499370, \n",
+    "'checkpoint_dir_name': None, 'done': True, 'training_iteration': 1, 'trial_id': '480ac050', 'date': '2025-11-18_20-56-10', 'time_this_iter_s': 2.0268948078155518, 'time_total_s': 2.0268948078155518, 'pid': 35181,\n",
+    "'hostname': 'Tobys-MacBook-Pro.local', 'node_ip': '127.0.0.1', 'config': {'component.controller1.arg.threshold': 10, 'component.controller2.arg.threshold': 38}, 'time_since_restore': 2.0268948078155518, \n",
+    "'iterations_since_restore': 1, 'experiment_tag': '23_component_controller1_arg_threshold=10,component_controller2_arg_threshold=38'}\n",
     "```\n"
    ]
   }
diff --git a/examples/tutorials/002_complex_processes/branching.py b/examples/tutorials/002_complex_processes/branching.py
@@ -12,7 +12,6 @@
 
 
 async def main() -> None:
-    # --8<-- [start:main]
     connect = lambda in_, out_: AsyncioConnector(  # (1)!
         spec=ConnectorSpec(source=in_, target=out_)
     )
diff --git a/examples/tutorials/002_complex_processes/components.py b/examples/tutorials/002_complex_processes/components.py
@@ -76,3 +76,12 @@ async def step(self) -> None:
     async def destroy(self) -> None:
         self._f.close()
 # --8<-- [end:components]
+
+# ---8<-- [start:parameters]
+class ScaleFromParameter(Component):
+    """Implements `x = a * scale` using parameters."""
+    io = IO(inputs=["a"], outputs=["x"])
+
+    async def step(self) -> None:
+        self.x = self.a * self.parameters.get("scale", 1)
+# ---8<-- [end:parameters]
diff --git a/examples/tutorials/002_complex_processes/loop.py b/examples/tutorials/002_complex_processes/loop.py
@@ -12,7 +12,6 @@
 
 
 async def main() -> None:
-    # --8<-- [start:main]
     connect = lambda in_, out_: AsyncioConnector(
         spec=ConnectorSpec(source=in_, target=out_)
     )
diff --git a/examples/tutorials/002_complex_processes/parameters.py b/examples/tutorials/002_complex_processes/parameters.py
@@ -0,0 +1,42 @@
+"""An example using process parameters."""
+
+# fmt: off
+import asyncio
+
+from plugboard.connector import AsyncioConnector
+from plugboard.process import LocalProcess
+from plugboard.schemas import ConnectorSpec
+
+from components import Random, Save, ScaleFromParameter, Sum
+
+
+async def main() -> None:
+    # --8<-- [start:main]
+    connect = lambda in_, out_: AsyncioConnector(
+        spec=ConnectorSpec(source=in_, target=out_)
+    )
+    process = LocalProcess(
+        components=[
+            Random(name="random", iters=5, low=0, high=10),
+            ScaleFromParameter(name="scale_a"),
+            ScaleFromParameter(name="scale_b", parameters={"scale": 2.0}),  # (2)!
+            Sum(name="sum"),
+            Save(name="save-input", path="input.txt"),
+            Save(name="save-output", path="output.txt"),
+        ],
+        connectors=[
+            connect("random.x", "save-input.value_to_save"),
+            connect("random.x", "scale_a.a"),
+            connect("random.x", "scale_b.a"),
+            connect("scale_a.x", "sum.a"),
+            connect("scale_b.x", "sum.b"),
+            connect("sum.x", "save-output.value_to_save"),
+        ],
+        parameters={"scale": 0.5},  # (1)!
+    )
+    async with process:
+        await process.run()
+    # --8<-- [end:main]
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/examples/tutorials/002_complex_processes/parameters.yaml b/examples/tutorials/002_complex_processes/parameters.yaml
@@ -0,0 +1,75 @@
+plugboard:
+  process:
+    args:
+      components:
+      - type: components.Random
+        args:
+          name: random
+          iters: 5
+          low: 0
+          high: 10
+      - type: components.ScaleFromParameter
+        args:
+          name: scale_a
+      - type: components.ScaleFromParameter
+        args:
+          name: scale_b
+          parameters:
+            scale: 2.0
+      - type: components.Sum
+        args:
+          name: sum
+      - type: components.Save
+        args:
+          name: save-input
+          path: input.txt
+      - type: components.Save
+        args:
+          name: save-output
+          path: output.txt
+      connectors:
+      - source:
+          entity: random
+          descriptor: x
+        target:
+          entity: save-input
+          descriptor: value_to_save
+        mode: pipeline
+      - source:
+          entity: random
+          descriptor: x
+        target:
+          entity: scale_a
+          descriptor: a
+        mode: pipeline
+      - source:
+          entity: random
+          descriptor: x
+        target:
+          entity: scale_b
+          descriptor: a
+        mode: pipeline
+      - source:
+          entity: scale_a
+          descriptor: x
+        target:
+          entity: sum
+          descriptor: a
+        mode: pipeline
+      - source:
+          entity: scale_b
+          descriptor: x
+        target:
+          entity: sum
+          descriptor: b
+        mode: pipeline
+      - source:
+          entity: sum
+          descriptor: x
+        target:
+          entity: save-output
+          descriptor: value_to_save
+        mode: pipeline
+      parameters:
+        scale: 0.5
+    type: plugboard.process.LocalProcess
diff --git a/examples/tutorials/006_optimisation/hello_tuner.py b/examples/tutorials/006_optimisation/hello_tuner.py
@@ -127,7 +127,9 @@ def custom_space(trial: Trial) -> dict[str, _t.Any] | None:
     )
     result = tuner.run(spec=process_spec)
     print(
-        f"Best parameters: angle={result.config['trajectory.angle']}, velocity={result.config['trajectory.velocity']}"
+        "Best parameters: "
+        f"angle={result.config['component.trajectory.arg.angle']}, "
+        f"velocity={result.config['component.trajectory.arg.velocity']}"
     )
-    print(f"Best max height: {result.metrics['max-height.max_y']}")
+    print(f"Best max height: {result.metrics['component.max-height.field.max_y']}")
     # --8<-- [end:run_tuner]
diff --git a/plugboard/component/component.py b/plugboard/component/component.py
@@ -53,7 +53,7 @@ def __init__(
         *,
         name: str,
         initial_values: _t.Optional[dict[str, _t.Iterable]] = None,
-        parameters: _t.Optional[dict] = None,
+        parameters: _t.Optional[dict[str, _t.Any]] = None,
         state: _t.Optional[StateBackend] = None,
         constraints: _t.Optional[dict] = None,
     ) -> None:
@@ -111,6 +111,11 @@ def status(self) -> Status:
         """Gets the status of the component."""
         return self._status
 
+    @property
+    def parameters(self) -> dict[str, _t.Any]:
+        """Gets the parameters of the component."""
+        return self._parameters
+
     @classmethod
     def _configure_io(cls) -> None:
         # Get all parent classes that are Component subclasses
@@ -221,6 +226,9 @@ async def connect_state(self, state: _t.Optional[StateBackend] = None) -> None:
         with self._job_id_ctx():
             await self._state.upsert_component(self)
             self._state_is_connected = True
+        # Merge this component's parameter values from those in the process
+        process = await self._state.get_process_for_component(self.id)
+        self._parameters = {**process["parameters"], **self._parameters}
 
     async def init(self) -> None:
         """Performs component initialisation actions."""
diff --git a/plugboard/process/process.py b/plugboard/process/process.py
@@ -28,7 +28,7 @@ def __init__(
         components: _t.Iterable[Component],
         connectors: _t.Iterable[Connector],
         name: _t.Optional[str] = None,
-        parameters: _t.Optional[dict] = None,
+        parameters: _t.Optional[dict[str, _t.Any]] = None,
         state: _t.Optional[StateBackend] = None,
     ) -> None:
         """Instantiates a `Process`.
@@ -37,13 +37,14 @@ def __init__(
             components: The components in the `Process`.
             connectors: The connectors between the components.
             name: Optional; Name for this `Process`.
-            parameters: Optional; Parameters for the `Process`.
+            parameters: Optional; Parameters for the `Process`. These will be shared across all
+                `Component` objects within the `Process`.
             state: Optional; `StateBackend` for the `Process`.
         """
         self.name = name or f"{self.__class__.__name__}_{gen_rand_str(8)}"
         self.components: dict[str, Component] = {c.id: c for c in components}
         self.connectors: dict[str, Connector] = {c.id: c for c in connectors}
-        self.parameters: dict = parameters or {}
+        self.parameters: dict[str, _t.Any] = parameters or {}
         self._state: StateBackend = state or self._default_state_cls()
         self._status = Status.CREATED
         self._state_is_connected: bool = False
diff --git a/plugboard/schemas/process.py b/plugboard/schemas/process.py
@@ -29,7 +29,8 @@ class ProcessArgsSpec(PlugboardBaseModel, extra="allow"):
         components: Specifies each `Component` in the `Process`.
         connectors: Specifies the connections between each `Component`.
         name: Unique identifier for `Process`.
-        parameters: Parameters for the `Process`.
+        parameters: Parameters for the `Process`. These will be shared across all `Component`
+            objects within the `Process`.
         state: Optional; Specifies the `StateBackend` used for the `Process`.
     """
 
diff --git a/plugboard/schemas/tune.py b/plugboard/schemas/tune.py
@@ -3,7 +3,7 @@
 from abc import ABC
 import typing as _t
 
-from pydantic import Field, PositiveInt, model_validator
+from pydantic import Field, PositiveInt, ValidationInfo, field_validator, model_validator
 
 from plugboard.schemas._common import PlugboardBaseModel
 
@@ -36,20 +36,39 @@ class BaseFieldSpec(PlugboardBaseModel, ABC):
 
     Attributes:
         object_type: The type of object on which the field is defined. Defaults to "component".
-        object_name: The name of the object on which the field is defined.
+        object_name: Optional; The name of the object on which the field is defined.
         field_type: The type of field. This can be "arg", "initial_value", or "field".
         field_name: The name of the field.
     """
 
-    object_type: _t.Literal["component"] = Field("component", exclude=True)
-    object_name: str = Field(..., exclude=True)
-    field_type: _t.Literal["arg", "initial_value", "field"] = Field(..., exclude=True)
+    object_type: _t.Literal["component", "process"] = Field("component", exclude=True)
+    object_name: str | None = Field(default=None, exclude=True)
+    field_type: _t.Literal["arg", "initial_value", "field", "parameter"] = Field(..., exclude=True)
     field_name: str = Field(..., exclude=True)
 
+    @field_validator("object_name")
+    @classmethod
+    def _validate_object_name(cls, v: str | None, info: ValidationInfo) -> str | None:
+        if info.data.get("object_type") == "component" and v is None:  # pragma: no cover
+            raise ValueError("Component name must be specified.")
+        return v
+
+    @field_validator("field_type")
+    @classmethod
+    def _validate_field_type(cls, v: str, info: ValidationInfo) -> str:
+        if info.data.get("object_type") == "process":  # pragma: no cover
+            if v != "parameter":
+                raise ValueError("If object_type is 'process', field_type must be 'parameter'.")
+        return v
+
     @property
     def full_name(self) -> str:
         """Returns the full name of the field, including the object name and field name."""
-        return f"{self.object_name}.{self.field_name}"
+        return (
+            f"{self.object_type}."
+            f"{self.object_name if self.object_name else 'default'}."
+            f"{self.field_type}.{self.field_name}"
+        )
 
 
 class ObjectiveSpec(BaseFieldSpec):
diff --git a/plugboard/tune/tune.py b/plugboard/tune/tune.py
@@ -167,21 +167,31 @@ def _build_parameter(
     def _override_parameter(
         process: ProcessSpec, param: ParameterSpec, value: _t.Any
     ) -> None:  # pragma: no cover
-        if param.object_type != "component":
-            raise NotImplementedError("Only component parameters are currently supported.")
-        try:
-            component = next(c for c in process.args.components if c.args.name == param.object_name)
-        except StopIteration:
-            raise ValueError(f"Component {param.object_name} not found in process.")
-        if param.field_type == "arg":
-            setattr(component.args, param.field_name, value)
-        elif param.field_type == "initial_value":
-            component.args.initial_values[param.field_name] = value
+        if param.object_type == "component":
+            try:
+                component = next(
+                    c for c in process.args.components if c.args.name == param.object_name
+                )
+            except StopIteration:
+                raise ValueError(f"Component {param.object_name} not found in process.")
+            if param.field_type == "arg":
+                setattr(component.args, param.field_name, value)
+            elif param.field_type == "initial_value":
+                component.args.initial_values[param.field_name] = value
+            elif param.field_type == "parameter":
+                component.args.parameters[param.field_name] = value
+        elif param.object_type == "process":
+            if param.field_type == "parameter":
+                process.args.parameters[param.field_name] = value
+        else:
+            raise ValueError(f"Unknown object type {param.object_type} for parameter override.")
 
     @staticmethod
     def _get_objective(process: Process, objective: ObjectiveSpec) -> _t.Any:  # pragma: no cover
         if objective.object_type != "component":
             raise NotImplementedError("Only component objectives are currently supported.")
+        if objective.object_name is None:
+            raise ValueError("Objective component name must be specified.")
         component = process.components[objective.object_name]
         return getattr(component, objective.field_name)
 
diff --git a/tests/integration/test_process_with_components_run.py b/tests/integration/test_process_with_components_run.py
@@ -50,14 +50,16 @@ async def step(self) -> None:
 
 
 class B(ComponentTestHelper):
-    io = IO(inputs=["in_1"], outputs=["out_1"])
+    io = IO(inputs=["in_1"], outputs=["out_1", "out_2"])
 
     def __init__(self, *args: _t.Any, factor: float = 1.0, **kwargs: _t.Any) -> None:
         super().__init__(*args, **kwargs)
         self._factor = factor
 
     async def step(self) -> None:
+        # Compute out_1 using factor argument, and out_2 using parameter 'factor'
         self.out_1 = self._factor * self.in_1
+        self.out_2 = self.parameters.get("factor", 1.0) * self.in_1
         await super().step()
 
 
diff --git a/tests/integration/test_state_backend.py b/tests/integration/test_state_backend.py
diff --git a/tests/integration/test_tuner.py b/tests/integration/test_tuner.py
diff --git a/tests/unit/test_tuner.py b/tests/unit/test_tuner.py

Original file line number	Diff line number	Diff line change
`@@ -12,7 +12,6 @@`
`12`	`12`
`13`	`13`
`14`	`14`	`async def main() -> None:`
`15`		`- # --8<-- [start:main]`
`16`	`15`	`connect = lambda in_, out_: AsyncioConnector( # (1)!`
`17`	`16`	`spec=ConnectorSpec(source=in_, target=out_)`
`18`	`17`	`)`
Original file line number	Diff line number	Diff line change
`@@ -127,7 +127,9 @@ def custom_space(trial: Trial) -> dict[str, _t.Any] \| None:`
`127`	`127`	`)`
`128`	`128`	`result = tuner.run(spec=process_spec)`
`129`	`129`	`print(`
`130`		`- f"Best parameters: angle={result.config['trajectory.angle']}, velocity={result.config['trajectory.velocity']}"`
	`130`	`+ "Best parameters: "`
	`131`	`+ f"angle={result.config['component.trajectory.arg.angle']}, "`
	`132`	`+ f"velocity={result.config['component.trajectory.arg.velocity']}"`
`131`	`133`	`)`
`132`		`- print(f"Best max height: {result.metrics['max-height.max_y']}")`
	`134`	`+ print(f"Best max height: {result.metrics['component.max-height.field.max_y']}")`
`133`	`135`	`# --8<-- [end:run_tuner]`