Recurrent Pipeline

Demonstrates RecurrentNode: gradient descent on (w - target)², unrolled into one node per step. Each step writes its own checkpoint file, and a final report node reads the last one.

Usage

cargo run --example recurrent_app -- run
cargo run --example recurrent_app -- run --from-nodes train/3
cargo run --example recurrent_app -- run --catalog checkpoints.count=20
cargo run --example recurrent_app -- check
cargo run --example recurrent_app -- viz

Types

#[derive(Serialize, Deserialize)]
pub struct Catalog {
    /// The starting point, read by iteration 0.
    pub init_weights: TextDataset,
    /// One checkpoint per iteration: `checkpoints.0`, `checkpoints.1`, …
    #[serde(deserialize_with = "pondrs::datasets::expand_indexed")]
    pub checkpoints: Vec<TextDataset>,
    pub report: JsonDataset,
}

#[derive(Serialize, Deserialize)]
pub struct Params {
    pub target: Param<f64>,
    pub learning_rate: Param<f64>,
}

checkpoints is a Vec<TextDataset> written as a count and a template:

checkpoints:
  count: 5
  template:
    path: "ckpt/epoch_{i}.txt"

Node function

/// One gradient step on `(w - target)^2`.
pub fn step(weights: String, target: f64, lr: f64) -> (String,) {
    let w: f64 = weights.trim().parse().unwrap();
    let grad = 2.0 * (w - target);
    ((w - lr * grad).to_string(),)
}

Pipeline

pub fn pipeline<'a>(cat: &'a Catalog, params: &'a Params) -> impl Steps<pondrs::error::PondError> + 'a {
    (
        RecurrentNode {
            name: "train",
            state: &cat.checkpoints,
            init: &cat.init_weights,
            input: |prev, _cur| (prev, &params.target, &params.learning_rate),
            output: |cur| (cur,),
            func: step,
        }
        .build(),
        Node {
            name: "report",
            input: (cat.checkpoints.last().expect("at least one iteration"),),
            output: (&cat.report,),
            func: |w: String| (json!({ "final_weights": w.trim().parse::<f64>().unwrap() }),),
        },
    )
}

App

fn main() -> Result<(), PondError> {
    let dir = data_dir();
    write_fixtures(&dir);

    pondrs::app::App::from_yaml(
        dir.join("catalog.yml").to_str().unwrap(),
        dir.join("params.yml").to_str().unwrap(),
    )?
    .with_hooks((
        LoggingHook::new(),
        VizHook::new("http://localhost:8080".to_string()),
    ))
    .with_args(std::env::args_os())?
    .dispatch(pipeline)
}

See Recurrent Nodes for how unrolling works.