Example: train a model, then predict with it
Synced from bicycle-studio-api
Source: studio_guide("example-train-predict") (GET /api/studio/v1/guide/example-train-predict) rendered at origin/platform (1afd83f), synced 2026-09-28. Do not edit this page here; change the source and run yarn sync:studio.
Build this when a forecast should come from a model you retrain on a schedule, and later steps should use the saved model rather than refit it.
Person-only stops: a person publishes demo_train and demo_predict in Studio (a first version always needs a
person). You then re-pin the published versions in the workflow and save. A person publishes the workflow.
Files
demo_train/function.json
{
"schema": "bicycle.function/v1",
"name": "demo_train",
"kind": "code",
"title": "Train a daily trend model",
"entrypoint": "main:handler",
"image": "bda-python:3",
"mode": "async",
"timeout_ms": 120000,
"resources": {
"class": "fn-small"
},
"input_schema": {
"type": "object",
"properties": {
"inputs": {
"type": "object"
},
"params": {
"type": "object"
},
"run": {
"type": "object"
}
}
},
"output_schema": {
"type": "object",
"required": [
"blob",
"sha256"
],
"properties": {
"blob": {
"type": "string"
},
"sha256": {
"type": "string"
},
"n_rows": {
"type": "integer"
},
"t_next": {
"type": "integer"
}
}
},
"capabilities": [
{
"id": "blob.write",
"names": [
"model-*",
"model-latest.json"
]
},
{
"id": "blob.read",
"names": [
"model-latest.json"
]
}
],
"visibility": {
"audience": "tenant",
"expose": {
"apps": false,
"workflows": true,
"agents": false,
"mcp": false
}
},
"docs": {
"summary": "Fits a Ridge trend on daily rows and saves model-<date>.skops plus a model-latest.json pointer."
},
"tests": [
{
"name": "trains-and-writes-pointer",
"input": {
"run": {
"logical_date": "2026-09-21"
},
"params": {
"model_blob": "model-2026-09-21.skops",
"latest_blob": "model-latest.json"
},
"inputs": {
"daily": [
{
"day": "2026-09-01",
"bookings": 5
},
{
"day": "2026-09-02",
"bookings": 7
},
{
"day": "2026-09-03",
"bookings": 9
},
{
"day": "2026-09-04",
"bookings": 11
},
{
"day": "2026-09-05",
"bookings": 13
},
{
"day": "2026-09-06",
"bookings": 15
},
{
"day": "2026-09-07",
"bookings": 17
},
{
"day": "2026-09-08",
"bookings": 19
},
{
"day": "2026-09-09",
"bookings": 21
},
{
"day": "2026-09-10",
"bookings": 23
}
]
}
},
"expect": {
"/blob": "model-2026-09-21.skops",
"/n_rows": 10,
"/t_next": 10,
"/r2": {
"$approx": 1.0,
"abs": 0.01
}
}
}
]
}
demo_train/main.py
import hashlib
import json
import numpy as np
import skops.io as sio
from sklearn.linear_model import Ridge
def rows_of(table):
"""A table input arrives as a list of row objects or as {columns, rows}."""
if isinstance(table, dict) and "rows" in table:
names = [c["name"] if isinstance(c, dict) else c for c in table.get("columns", [])]
return [dict(zip(names, r)) if isinstance(r, list) else r for r in table["rows"]]
return list(table or [])
def handler(input, ctx):
params, run = input.get("params") or {}, input.get("run") or {}
target = params.get("target", "bookings")
rows = sorted(rows_of((input.get("inputs") or {}).get("daily")), key=lambda r: str(r["day"]))
if len(rows) < 3:
raise ValueError(f"need at least 3 rows to train, got {len(rows)}")
X = np.arange(len(rows), dtype=float).reshape(-1, 1) # feature t = day index
y = np.array([float(r[target]) for r in rows])
model = Ridge(alpha=1.0).fit(X, y)
data = sio.dumps(model)
previous = ctx.blob.get(params["latest_blob"])
latest = {
"blob": params["model_blob"],
"sha256": hashlib.sha256(data).hexdigest(),
"logical_date": run.get("logical_date"),
"features": ["t"],
"t_next": len(rows),
"n_rows": len(rows),
"last_day": str(rows[-1]["day"])[:10],
"r2": round(float(model.score(X, y)), 4),
"previous_sha256": json.loads(previous)["sha256"] if previous else None,
}
ctx.blob.put(params["model_blob"], data, "application/octet-stream")
ctx.blob.put(params["latest_blob"], json.dumps(latest), "application/json")
return latest
demo_predict/function.json
{
"schema": "bicycle.function/v1",
"name": "demo_predict",
"kind": "code",
"title": "Predict from the latest model",
"entrypoint": "main:handler",
"image": "bda-python:3",
"mode": "async",
"timeout_ms": 60000,
"resources": {
"class": "fn-xs"
},
"input_schema": {
"type": "object",
"properties": {
"inputs": {
"type": "object"
},
"params": {
"type": "object"
},
"run": {
"type": "object"
}
}
},
"output_schema": {
"type": "object",
"required": [
"status",
"predictions"
],
"properties": {
"status": {
"type": "string"
},
"predictions": {
"type": "array"
},
"model_blob": {
"type": "string"
},
"sha256": {
"type": "string"
}
}
},
"capabilities": [
{
"id": "blob.read",
"names": [
"model-*",
"model-latest.json"
]
}
],
"visibility": {
"audience": "tenant",
"expose": {
"apps": false,
"workflows": true,
"agents": false,
"mcp": false
}
},
"docs": {
"summary": "Checks the model's sha256 against model-latest.json, loads it with skops trusted types and predicts the next days."
},
"tests": [
{
"name": "predicts-with-a-real-model",
"input": {
"params": {
"latest_blob": "model-latest.json",
"horizon_days": 3
}
},
"fixtures": {
"blob.read:model-latest.json": "fixtures/latest.json",
"blob.read:model-2026-09-21.skops": "fixtures/model.skops"
},
"expect": {
"/status": "ok",
"/predictions/0/predicted": {
"$approx": 84.99,
"abs": 0.05
}
}
},
{
"name": "no-model-yet",
"input": {
"params": {
"latest_blob": "model-latest.json"
}
},
"expect": {
"/status": "no_model"
}
}
]
}
demo_predict/main.py
import hashlib
import json
import skops.io as sio
TRUSTED = {"sklearn.linear_model._ridge.Ridge"} # beyond skops' defaults; never pickle
def handler(input, ctx):
params = input.get("params") or {}
raw = ctx.blob.get(params.get("latest_blob", "model-latest.json"))
if raw is None:
return {"status": "no_model", "predictions": []}
latest = json.loads(raw)
data = ctx.blob.get(latest["blob"])
if data is None or hashlib.sha256(data).hexdigest() != latest["sha256"]:
raise ValueError("the model does not match model-latest.json")
untrusted = set(sio.get_untrusted_types(data=data))
if untrusted - TRUSTED:
raise ValueError(f"the model holds types outside the allowlist: {sorted(untrusted - TRUSTED)}")
model = sio.loads(data, trusted=sorted(untrusted))
t0, n = int(latest.get("t_next", 0)), int(params.get("horizon_days", 7))
yhat = model.predict([[float(t)] for t in range(t0, t0 + n)])
return {
"status": "ok",
"model_blob": latest["blob"],
"sha256": latest["sha256"],
"predictions": [{"step": i + 1, "t": t0 + i, "predicted": round(float(v), 2)} for i, v in enumerate(yhat)],
}
demo_predict/fixtures/model.skops is a real model, made with the image's versions (function_kinds:
scikit-learn==1.9.1, skops==0.16.0, numpy==2.5.3). fixtures/latest.json is
{"blob": "model-2026-09-21.skops", "sha256": "<sha256>", "t_next": 40}.
import base64, hashlib, numpy as np, skops.io as sio
from sklearn.linear_model import Ridge
X = np.arange(40, dtype=float).reshape(-1, 1)
data = sio.dumps(Ridge(alpha=1.0).fit(X, 2 * X[:, 0] + 5)) # predicts 84.99 at t = 40
print(hashlib.sha256(data).hexdigest(), base64.b64encode(data).decode())
workflow.json
{
"schema": "bicycle.workflow/v1",
"title": "Train weekly, predict the next days",
"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"}],
"queries": {
"daily_bookings": {"sql": "SELECT DATE_TRUNC('day', timestamp) AS day, bookings FROM Booking WHERE timestamp >= $from AND timestamp < $to GROUP BY 1 ORDER BY 1", "parameters": [{"name": "from", "type": "date"}, {"name": "to", "type": "date"}], "columns": [{"name": "day", "type": "date"}, {"name": "bookings", "type": "number"}], "maxLimit": 400}
},
"artifacts": {
"daily": {"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": "daily_bookings", "params": {"from": "${run.logical_date - P56D}", "to": "${run.logical_date}"}}, "outputs": {"rows": "daily"}},
"train": {"kind": "function", "config": {"ref": "fn:<tenant>/demo_train@1", "params": {"model_blob": "model-${run.logical_date}.skops", "latest_blob": "model-latest.json"}}, "inputs": {"daily": "daily"}, "outputs": {"value": "trained"}},
"predict": {"kind": "function", "config": {"ref": "fn:<tenant>/demo_predict@1", "params": {"latest_blob": "model-latest.json", "horizon_days": 7}}, "inputs": {"trained": "trained"}, "outputs": {"predictions": "predictions", "value": "prediction_summary"}}
},
"triggers": {
"weekly": {"type": "cron", "cron": "0 6 * * 1", "targets": ["prediction_summary"]},
"manual": {"type": "manual"}
}
}
schemas/trained.json
{"type": "object", "required": ["blob", "sha256"], "properties": {"blob": {"type": "string"}, "sha256": {"type": "string"}, "n_rows": {"type": "integer"}, "logical_date": {"type": "string"}}}
schemas/prediction_summary.json
{"type": "object", "required": ["status"], "properties": {"status": {"type": "string"}, "model_blob": {"type": "string"}, "sha256": {"type": "string"}}}
A json artifact needs a schema file; without one the document does not validate.
Calls, in order
function_kinds(): the image's exact package versions. Build the fixture with them.function_create("demo_train", title)andfunction_create("demo_predict", title).function_put_files("demo_train", {"function.json": ..., "main.py": ...}): version 1,validated.function_put_files("demo_predict", {..., "fixtures/latest.json": ..., "fixtures/model.skops": {"content": "\<base64\>", "encoding": "base64"}}): version 1,validated, one version for all four files.function_test("demo_train", 1)andfunction_test("demo_predict", 1): every test passed, statetested. The predict test loads the real model and returns 84.99.function_try("demo_predict", input, n=1, blobs={"model-latest.json": ..., "model-2026-09-21.skops": {"content": "\<base64\>", "encoding": "base64"}}):status: okfrom a throwaway store seeded with those blobs.workflow_create(title, document, files, model="<model>"): valid, with afunction_draftwarning per pin.workflow_validate/workflow_plan: a try run would start; a publish refuses withfunction_unpublished.workflow_run(workflow_id, revision=1, logical_date="2026-09-22"): a try run. Its blobs are its own.workflow_run_describe(workflow_id, run_id):fetch56 rows,trainandpredictsucceeded, each with its invocation and "draft version (tested)".workflow_artifact(workflow_id, "predictions", version_id): seven rowsstep, t, predicted. Take theversion_idfrom the run's root invocation (invocation_get) when the describe table shortens it.
Stop here. Ask the person to publish both functions, re-pin the published versions, then ask them to publish the workflow.
Switch it off
workflow_disable(workflow_id, reason="...")
function_disable("demo_train", reason="...")
function_disable("demo_predict", reason="...")
Verified on preview 2026-09-28: function_kinds, function_create, function_put_files, function_test, function_try, workflow_create, workflow_validate, workflow_plan, workflow_run (try), workflow_run_describe, invocation_get, workflow_artifact, workflow_disable, function_disable ran; invocation ids inv_01M3J6WYW00ACG4CPKYN0AEQ1W (try), inv_01M3J6YESSPMTZQYTJ0P54S9RW (run), inv_01M3J6YNDSQKP9S209EDPED6QP (train), inv_01M3J6YW5J3E49W6P6YVRTHSTK (predict).
guide_version bb2461328de4