Train a model weekly, predict daily

This example keeps something between runs. A workflow trains a small statistical model on the last eight weeks of failed orders, saves the model file, and every day uses the latest saved model to predict tomorrow's failed orders per reason. A Predict button in the KPI app starts that workflow and shows the prediction. Nothing is emailed.

The new idea is workflow blobs: files a workflow keeps at its own address, which the functions it calls may read and write by name. That is where the model lives.

What you will have at the end

• A workflow Train weekly, predict daily with two steps, train and predict, and two schedules: train every Monday at 06:00, predict every day at 06:00.
• Two saved files inside the workflow: model-2026-09-21.skops (one per training date) and model-latest.json, a small pointer saying which model is current and what its checksum is.
• A Predict button on Test KPI app that shows, for example: "Tomorrow: 38 failed orders expected (card_declined 22, out_of_stock 9, other 7), from the model trained on 21 September". Check any such number against what you know: a prediction is only as good as the eight weeks it saw.

What the agent will ask you before it builds

1. What exactly should be predicted? Tomorrow's failed orders per reason, from the last eight weeks by day and reason. The agent shows you those eight weeks' totals first: "The last 8 weeks have 4,912 failed orders; card_declined is 61 %. Does that match?"
2. How often should the model be retrained, and how old may it be? Weekly here, and a model is reused while it is under seven days old, so the daily predict never retrains.
3. What kind of model? The agent should propose something small and honest for eight weeks of daily counts (a per-reason linear trend or a moving average from scikit-learn) and say plainly how rough it will be. If you need a real forecast, that is a different project; this example is about the plumbing.
4. Where should the prediction be seen? In the app, on a button. The agent should say why the app calls the workflow and not the predict function directly: the model files belong to the workflow, and only a run of that workflow may read them.
5. Safety of the model file. The agent should say, unprompted, that it saves the model with skops and loads it only with an explicit list of allowed types, never with pickle (a pickle file can run arbitrary code when loaded), and that it checks the file's sha256 checksum against the pointer before loading.

The pieces it creates

Piece	Name	What it is
Code function	ml_train (class fn-small)	Reads the history, fits a model, writes model-<date>.skops and model-latest.json; reuses a recent model instead
Code function	ml_predict	Reads the pointer, checks the checksum, loads the model with a type allowlist, scores tomorrow
Workflow	Train weekly, predict daily	Declares the query, the two blobs, the two function steps, two crons
App version	Test KPI app v3	The Predict button and the import wf:<tenant>/train_predict@1

Steps, in order

1. Prove the query (agent)

query_run for failed orders by day and reason over the last eight weeks, shown to you as totals to check.

2. Write the two functions (agent)

Both are kind: code on bda-python:3. What matters is in function.json:

• ml_train: resources.class: fn-small (training needs more room than fn-xs), capabilities blob.read ["model-latest.json"] and blob.write ["model-*", "model-latest.json"].
• ml_predict: capabilities blob.read ["model-*", "model-latest.json"]. It cannot write anything.

Neither lists semantic.query: the history rows arrive from the workflow's query step as an input table. Each has at least one test with a fixture.

# ml_train main.py (outline; the agent writes the full version)
import hashlib, json
import skops.io

FEATURES = ["day_of_week", "reason"]

def handler(input, ctx):
    params, run = input["params"], input["run"]
    raw = ctx.blob.get(params["latest_blob"])
    if raw and days_between(run["logical_date"], json.loads(raw)["logical_date"]) < params["max_age_days"]:
        return {**json.loads(raw), "reused": True}          # a recent model: do not retrain
    history = input["inputs"]["history"]                     # the fetch step's rows
    model = fit(history)                                      # scikit-learn, small and honest
    data = skops.io.dumps(model)
    latest = {"blob": params["model_blob"], "sha256": hashlib.sha256(data).hexdigest(),
              "logical_date": run["logical_date"], "features": FEATURES}
    ctx.blob.put(params["model_blob"], data, "application/octet-stream")
    ctx.blob.put(params["latest_blob"], json.dumps(latest), "application/json")
    return latest

# ml_predict main.py (outline)
import hashlib, json
import skops.io

TRUSTED = ["sklearn.linear_model._base.LinearRegression", "numpy.dtype"]   # only what the model needs

def handler(input, ctx):
    latest = json.loads(ctx.blob.get(input["params"]["latest_blob"]))
    data = ctx.blob.get(latest["blob"])
    if hashlib.sha256(data).hexdigest() != latest["sha256"]:
        raise ValueError("the model does not match model-latest.json")
    if set(skops.io.get_untrusted_types(data=data)) - set(TRUSTED):
        raise ValueError("the model holds types outside the allowlist")
    model = skops.io.loads(data, trusted=TRUSTED)
    rows = score(model, input["inputs"]["history"])           # one row per reason
    return {"predictions": rows, "n": sum(r["expected"] for r in rows),
            "model_blob": latest["blob"], "sha256": latest["sha256"]}

Three things to check the agent did:

• The model is saved with skops.io.dumps and loaded with skops.io.loads(..., trusted=TRUSTED) after checking get_untrusted_types. There is no pickle anywhere.
• The checksum in model-latest.json is compared with the file before loading.
• Blobs are not versioned for you, so the date is in the file name and the pointer file says which one is current.

The agent runs function_test on both; the tests should report tested. The function image ships numpy, pandas, pyarrow, DuckDB, scikit-learn, joblib and skops, so both import; there is no pip at run time, so anything else is not available.

3. Publish both functions: Only a person

function_publish on each first version answers with a link. Only a person: confirm both. Read what each may do: train may write two named blobs; predict may only read them. The agent notes the pinned references fn:<tenant>/ml_train@1 and fn:<tenant>/ml_predict@1.

4. Write the workflow (agent)

{
  "schema": "bicycle.workflow/v1",
  "title": "Train weekly, predict daily",
  "queries": {
    "failed_by_day_reason": {
      "sql": "SELECT day, reason, failed_orders FROM orders WHERE day >= $from AND day < $to ORDER BY day",
      "parameters": [{"name": "from", "type": "date"}, {"name": "to", "type": "date"}],
      "columns": [{"name": "day", "type": "date"}, {"name": "reason", "type": "string"},
                  {"name": "failed_orders", "type": "number"}],
      "maxLimit": 10000
    }
  },
  "blobs": [
    {"name": "model-*", "kind": "binary", "max_bytes": 20971520, "purpose": "one trained model per logical date"},
    {"name": "model-latest.json", "kind": "json", "max_bytes": 65536, "purpose": "which model is current"}
  ],
  "artifacts": {
    "history": {"type": "table"},
    "trained": {"type": "json", "schema": "schemas/trained.json"},
    "predictions": {"type": "table"},
    "prediction_summary": {"type": "json", "schema": "schemas/prediction_summary.json"}
  },
  "nodes": {
    "fetch": {"kind": "query",
      "config": {"query": "failed_by_day_reason",
                 "params": {"from": "${run.logical_date - P56D}", "to": "${run.logical_date + P1D}"}},
      "outputs": {"rows": "history"}},
    "train": {"kind": "function",
      "config": {"ref": "fn:<tenant>/ml_train@1",
                 "params": {"model_blob": "model-${run.logical_date}.skops",
                            "latest_blob": "model-latest.json", "max_age_days": 7}},
      "inputs": {"history": "history"},
      "outputs": {"value": "trained"}},
    "predict": {"kind": "function",
      "config": {"ref": "fn:<tenant>/ml_predict@1", "params": {"latest_blob": "model-latest.json"}},
      "inputs": {"trained": "trained", "history": "history"},
      "outputs": {"predictions": "predictions", "value": "prediction_summary"}}
  },
  "triggers": {
    "weekly_train": {"type": "cron", "cron": "0 6 * * 1", "targets": ["trained"]},
    "daily_predict": {"type": "cron", "cron": "0 6 * * *", "targets": ["prediction_summary"]},
    "manual": {"type": "manual"}
  }
}

In words:

• The query is declared in the workflow, in its queries block, because the workflow is the deployment that runs it (the same rule as the app in example 1).
• blobs declares the two files by name pattern. A function may only touch a blob the workflow declares and its own function.json asks for.
• train takes the history and reuses the current model while it is under seven days old; predict waits for train (its output is an input) so a fresh model is always used the day it is trained.
• weekly_train targets trained; daily_predict targets prediction_summary, which also runs train, which reuses the model on six days out of seven. A P56D offset is eight weeks.

Nothing here sends anything: no action step, no email, no approval.

5. Try run, then publish (agent, then Your call)

workflow_patch with dry_run until valid, workflow_plan, then a try run of the draft. workflow_run_describe should show fetch, train and predict all done, trained with the model's name and checksum, predictions with one row per reason. A second try run the same week should show train reporting reused: true and finishing at once.

The workflow has no send step, but its first revision adds a query and a model, and those are gated changes, so workflow_publish answers that a person must publish it. Only a person: the agent gives you the workflow's link; you open it, read the review and press Publish. The pinned reference is wf:<tenant>/train_predict@1.

6. The Predict button (agent)

A new app version imports the workflow, by its wf: reference:

"functions": {
  "order_kpis": {"ref": "fn:<tenant>/order_kpis@1"},
  "train_predict": {"ref": "wf:<tenant>/train_predict@1"}
}

Not the predict function. The model files belong to the workflow, and only a run of that workflow may read them; a direct call of ml_predict from the app would have no blobs to read.

predictButton.onClick = async () => {
  predictButton.disabled = true
  const started = await bda.fn.call('train_predict', { targets: ['prediction_summary'] }, { wait: false })
  cancelButton.onClick = () => bda.fn.cancel(started.invocation_id)
  const watch = bda.fn.watch(started.invocation_id, batch => showProgress(batch.items))
  // e.g. "fetch succeeded · 392 rows · 1.1 s", "train succeeded (reused)", "predict succeeded"
  const out = bda.fn.outputOf(await watch.done)
  const summary = out.artifacts.prediction_summary.data     // { n, model_blob, sha256, ... }
  showPrediction(summary)                                   // as text
  if (out.partial) showNote(`Some steps did not finish: ${(out.failed ?? []).join(', ')}`)
}

Two rules that differ from calling a function:

• reuse does not apply to workflows. A workflow call is always a real run of the published revision, as the viewer. So the button never runs on page load; it runs when pressed. (A run for the same date that is already active is joined rather than started again, so a second press is cheap.)
• The output is the run's receipt: status, the artifacts it produced (with their data inline while small), and any send receipts (none here).

7. Publish the app version: Only a person

The version adds an import (train_predict), so a person reviews it. Only a person: open the link and confirm.

How to test it

1. Press Predict in the app. Watch the progress lines; the whole run takes seconds to a minute. Read the prediction and check it is in a plausible range against the eight-week totals from step 1.
2. Press Predict again. The run should be faster and the trace should say train reused the model.
3. Open Runs. The workflow run has one entry per step; under train and predict are the function runs themselves, with their own timing. Nothing shows Waiting for approval, because nothing sends.
4. In Studio, open the workflow and look at its artifacts: trained (the model's name and checksum), predictions (rows), and the two blobs (the agent's tool for reading artifacts is landing; today they are on the workflow's page).
5. Check your inbox: nothing arrived. This workflow sends nothing.

How to switch it off

• Pause both schedules after testing: in Studio, open the workflow and choose Pause on weekly_train and daily_predict. Otherwise it trains every Monday and predicts every morning, at a small compute cost each time.
• Disable the workflow on its page when you are done; the app's Predict button then fails in words (workflow_disabled), so publish an app version without the button too.
• Disable the two functions on their pages, once the workflow is disabled.

What you learned

• A workflow keeps files between runs (blobs), declared by name; functions ask for them by name.
• Model files are saved with skops and loaded with a type allowlist and a checksum check, never pickle.
• An app that needs a workflow's files imports the workflow, not the function; reuse does not apply, so it runs on a button, never on load.
• Two crons on one workflow, one per target.

Read more: what a workflow keeps between runs, calling a workflow from an app, schedules and failures, limits and safety for blob and function sizes.

Next: What changed, and why: Detect and Explain.

Copy for your agent

On preview, train a model weekly that predicts tomorrow's failed orders per reason from the last
8 weeks, keep the latest model, predict daily, and add a "Predict" button to my "Test KPI app"
(new version) that shows the prediction. Email nothing.

How to work with me:
- Follow the guide's train/predict pattern: two code functions, ml_train (fn-small, blob.write on
  model-* and model-latest.json) and ml_predict (blob.read), skops with an explicit type allowlist,
  never pickle, a sha256 check against model-latest.json, and the blobs declared in the workflow.
- Declare the history query in the workflow's queries block. Run it first and show me the 8-week
  totals to check.
- Tell me before you build what kind of model you propose and how rough it will be.
- Steps that need me: confirming each function's first publish (give me the links), and the app
  version that adds the workflow import. Ask me before publishing the workflow.
- Try run first; show me the trained and predictions artifacts; run again and show me the model
  was reused.
- The app imports the WORKFLOW by wf: ref, not the predict function, and you tell me why. Start it
  with wait: false from the button's onClick, watch progress, allow Cancel, show
  prediction_summary from outputOf(final) as text. Do not use reuse; never run it on page load.
- Two crons: train Monday 06:00, predict daily 06:00. At the end tell me how to pause both in
  Studio and how to disable everything.
- Only create things titled "Test ...". Never change an existing app.