SR-Forge

SequentialModel

Most real-world deep learning pipelines are not a single model. You might encode first, then transform, then decode. Or split data into branches, process them separately, and merge. SequentialModel lets you compose these multi-stage pipelines declaratively using a simple flow DSL — no glue code needed.

You've already learned about the building blocks — Models, DataTransforms, and EntryTransforms. SequentialModel is how you wire them together into a pipeline, using IO binding to route data between stages automatically.

---

Quick Reference

Every flow line follows the same pattern:

<inputs> -> <module> -> <outputs>

• Inputs: Entry field names that the module reads
• Module: A named Model, EntryTransform, or DataTransform
• Outputs: Entry field names that the module writes

Minimal Example

from srforge.models import SequentialModel, Model
from srforge.data import Entry

class Encoder(Model):
    def forward(self, image):
        return self.net(image)

class Decoder(Model):
    def forward(self, features):
        return self.net(features)

seq = SequentialModel(
    modules={"encode": Encoder(), "decode": Decoder()},
    flow="""
        x -> encode -> features
        features -> decode -> y
    """,
)

entry = Entry(x=some_tensor)
result = seq(entry)
# result.features — intermediate result preserved
# result.y — final output

Module Types in Flow

Module Type	IO Source	Named Input Mapping	Auto-recurse into dicts/lists
Model	Flow DSL	Yes ((image=x) -> m)	No
EntryTransform	Constructor _key params	N/A (opaque)	No
DataTransform	Flow DSL	Yes ((data=x) -> t)	Yes (annotation-driven)

---

How It Works

A SequentialModel is itself a Model. It takes an Entry, runs it through a sequence of steps, and returns the modified Entry. Fields accumulate as the pipeline runs — step 2 can read fields produced by step 1.

Entry -> [step 1] -> Entry' -> [step 2] -> Entry'' -> ... -> Final Entry

Defining a SequentialModel

seq = SequentialModel(
    modules={"module_name": module_instance, ...},
    flow="...flow lines...",
)

The flow parameter can be a multiline string or a list of strings. The modules dict provides the named modules referenced in the flow.

---

Flow DSL Syntax

Basic Syntax

Each line has exactly three segments separated by ->:

<LHS> -> <module> -> <RHS>

1. LHS — Entry fields to read (positional or named)
2. Module segment — which module to execute
3. RHS — Entry fields to write

Writing Flow Lines

Multiline string (most readable):

flow = """
    # This is a comment
    x -> encoder -> features
    features -> decoder -> output
"""

List of strings:

flow = [
    "x -> encoder -> features",
    "features -> decoder -> output",
]

Multiple Inputs and Outputs

x, y -> module -> a, b
(x, y) -> module -> (a, b)

Both forms are equivalent. Parentheses are optional. This syntax works identically for Models and DataTransforms — inputs map to method parameters by signature order, outputs map to return values positionally.

Empty Inputs or Outputs

 -> set_attr -> y          # no inputs (writes only)
x -> remove -> ()          # no outputs (reads and deletes)

---

Models in Flow

Models are the most flexible module type. The flow line controls which Entry fields map to which parameters.

Positional Mapping

Fields map to forward() parameters by signature order. Outputs are assigned positionally to return values:

class TwoInputModel(Model):
    def forward(self, image, guide):
        output, conf = self.net(image, guide)
        return output, conf

flow = ["(x, edges) -> enhance -> (result, conf)"]
# "image" <- entry["x"]       (1st forward() param)
# "guide" <- entry["edges"]   (2nd forward() param)
# 1st return value -> entry["result"]
# 2nd return value -> entry["conf"]

Named Input Mapping

For clarity or different ordering, use named mapping on the LHS. Names match forward() parameter names:

class TwoInputModel(Model):
    def forward(self, image, guide):
        output, conf = self.net(image, guide)
        return output, conf

flow = ["(image=x, guide=edges) -> enhance -> (result, conf)"]
# "image" <- entry["x"], "guide" <- entry["edges"]
# 1st return value → entry["result"], 2nd → entry["conf"]

Each segment has one job: LHS = inputs, module = name, RHS = outputs.

Reusing the Same Model

A model can appear in multiple flow lines with shared weights:

encoder = Encoder()
seq = SequentialModel(
    modules={"encoder": encoder},
    flow="""
        x -> encoder -> h1
        h1 -> encoder -> h2
    """,
)
# encoder applied twice with different IO mappings, shared parameters

---

EntryTransforms in Flow

EntryTransforms are opaque steps in the flow DSL — field mapping is handled by the constructor's _key parameters, not the flow line. The flow just marks when to run the transform.

Basic Usage

from srforge.transform.entry import CopyFields

seq = SequentialModel(
    modules={"copy": CopyFields(field_key="y", output_key="z")},
    flow=[" -> copy -> "],
)

Different Field Mappings

Use separate instances for different field mappings:

copy1 = CopyFields(field_key="y", output_key="z")
copy2 = CopyFields(field_key="z", output_key="w")
seq = SequentialModel(
    modules={"copy1": copy1, "copy2": copy2},
    flow=[" -> copy1 -> ", " -> copy2 -> "],
)

Transforms with No Inputs or Outputs

Some transforms only write (e.g., SetAttribute), others only read and delete (e.g., RemoveFields):

from srforge.transform.entry import SetAttribute, RemoveFields

seq = SequentialModel(
    modules={
        "set": SetAttribute(5, attr_key="flag"),
        "rm": RemoveFields(field_key="temp"),
    },
    flow=[" -> set -> ", " -> rm -> "],
)

---

DataTransforms in Flow

DataTransforms are the simplest to use in a flow. They don't need pre-binding — the flow line becomes the binding.

Auto-Binding from Flow

from srforge.transform.data import Multiply

mul = Multiply(2.0)
seq = SequentialModel(
    modules={"mul": mul},
    flow=["x -> mul -> y"],
)

In-Place Transform

flow = ["x -> mul -> x"]   # entry["x"] overwritten with transformed result

Multiple Fields

To apply the same transform to multiple fields, use separate flow lines:

flow = [
    "x -> mul -> x2",
    "y -> mul -> y2",
]
# entry["x2"] = mul(entry["x"])
# entry["y2"] = mul(entry["y"])

Per-Sample DataTransforms

DataTransforms that implement transform_unbatched() instead of transform() work transparently in flows — no special syntax needed. The transform handles its own dispatch (batched vs per-sample) internally:

class PerSampleDouble(DataTransform):
    def transform_unbatched(self, image: torch.Tensor) -> torch.Tensor:
        return image * 2  # receives [C,H,W], no batch dim

seq = SequentialModel(
    modules={"double": PerSampleDouble()},
    flow="image -> double -> image",
)
# Each sample is doubled independently, then re-batched

Recursive on Nested Structures

If a field contains a nested dict or list, the DataTransform is applied recursively to every leaf tensor (when the transform parameter is Tensor-annotated):

entry = Entry(x={"a": tensor_a, "b": tensor_b})
flow = ["x -> mul -> y"]
# entry["y"] = {"a": mul(tensor_a), "b": mul(tensor_b)}

Named Input Mapping

For multi-parameter transforms, use named mapping on the LHS to specify which Entry field maps to which transform() parameter:

# MatchReference: transform(self, image: Tensor, reference: Tensor)
flow = ["(image=x, reference=ref) -> match -> matched"]
# image <- entry["x"], reference <- entry["ref"]
# result → entry["matched"]

Optional parameters can be omitted — the default value applies:

# transform(self, image: Tensor, mask: Tensor = None)
flow = ["(image=x) -> noise -> noisy"]           # mask=None
flow = ["(image=x, mask=m) -> noise -> noisy"]   # mask=entry["m"]

Reusability

DataTransforms are fully reusable — the same instance can appear in multiple flow lines with different fields. Instance state is never mutated; applications are computed externally:

mul = Multiply(2.0)
flow = [
    "x -> mul -> x_scaled",
    "y -> mul -> y_scaled",   # Same instance, different fields
]

See DataTransform: In SequentialModel for details on recursion and multi-input behavior.

---

Building Pipelines

Two-Stage Pipeline

seq = SequentialModel(
    modules={"stage1": FirstStage(), "stage2": SecondStage()},
    flow="""
        x -> stage1 -> intermediate
        intermediate -> stage2 -> y
    """,
)

Model + Transform Pipeline

from srforge.transform.data import Multiply
from srforge.transform.entry import CopyFields

seq = SequentialModel(
    modules={
        "m1": HRPassthrough(),
        "copy": CopyFields(field_key="y", output_key="z"),
        "scale": Multiply(2.0),
    },
    flow="""
        x -> m1 -> y
         -> copy ->
        z -> scale -> out
    """,
)

Multi-Input Model with Auxiliary Output

Outputs are mapped positionally — the first return value goes to the first output field, the second to the second, etc.:

class FusionModel(Model):
    def forward(self, image, guide):
        output, attn = self.fusion_net(image, guide)
        return output, attn

seq = SequentialModel(
    modules={"fuse": FusionModel()},
    flow=["(x, edges) -> fuse -> (result, attn)"],
)
# 1st return value → entry["result"]
# 2nd return value → entry["attn"]

---

Required Input Inference

SequentialModel automatically determines which Entry fields must be present before the pipeline runs:

flow = """
    x -> encoder -> features       # consumes "x" (not produced by any prior step)
    features -> decoder -> out     # consumes "features" (produced by step 1)
"""
# Required inputs: ["x"]

---

Field Overwrite Protection

Model steps cannot overwrite existing Entry fields:

entry = Entry(x=tensor_a, y=tensor_b)
seq(entry)  # KeyError: attempted to overwrite existing entry fields: ['y']

DataTransforms are an exception — they can overwrite in-place when input and output are the same field.

---

YAML Configuration

SequentialModel is fully configurable from YAML:

model:
  _target: srforge.models.SequentialModel
  params:
    modules:
      encoder:
        _target: MyModel
        params:
          hidden_dim: 128
      copy:
        _target: srforge.transform.entry.CopyFields
        params:
          field_key: features
          output_key: features_backup
      scale:
        _target: srforge.transform.data.Multiply
        params:
          value: 1.5
    flow:
      - "x -> encoder -> features"
      - " -> copy -> "
      - "features -> scale -> output"

Models and DataTransforms get IO from the flow DSL. EntryTransforms get field keys from their params: section — no io: key needed.

---

Common Errors

"Module 'xxx' not found in modules"

The module name in the flow must match a key in the modules dict.

"Module 'xxx' expects N inputs, got M"

Provide all required inputs in the LHS:

# Model needs 2 inputs
flow = ["x -> m1 -> y"]  # ERROR: only 1 input
flow = ["(x, g) -> m1 -> y"]  # OK

"DataTransform step 'xxx' requires flow inputs/outputs"

DataTransforms need at least an input field in the flow.

"Duplicate outputs in flow line"

Each output field name must be unique within a flow line.

---

Summary

	Model	EntryTransform	DataTransform
IO binding	Flow DSL	Constructor _key params	Flow DSL
Input mapping	Positional or named	N/A (opaque)	Positional or named
Output mapping	Positional only	N/A (opaque)	Positional only
Named syntax on LHS	Yes: (image=x) -> m1	No	Yes: (data=x) -> t
Auto-recurse into dicts/lists	No	No	Yes (annotation-driven)
Reusable (different mapping)	Yes	Separate instances	Yes

When to Use SequentialModel

• Multi-stage pipelines (encode -> transform -> decode)
• Mixing models and transforms in one forward pass
• Pipelines where intermediate results need to be preserved
• Declarative experiment configuration in YAML

When NOT to Use SequentialModel

• Single model, no transforms — just use the model directly
• Pure preprocessing — apply transforms directly in the dataset's transforms list
• Dynamic branching (if/else logic) — SequentialModel is strictly sequential

---

Next: Losses — Evaluation metrics and loss functions