Workflows API

A workflow is a small graph of typed blocks that turns a source (an image, a video, or a Pictograph dataset) into an answer: run a model over every frame, keep the classes you care about, optionally track and count them, and emit the results to a sink (JSON, CSV, a webhook, or an annotated video). You build the graph visually in the app, then drive runs from here.

A workflow’s model block runs your trained model directly, so no deployment is required. Runs are billed per frame processed (denser frame sampling costs more), charged as a deposit at start and trued up when the run finishes.

Every example below shows the Python SDK call and the equivalent raw REST request. The REST examples authenticate with an X-API-Key header; set PICTOGRAPH_API_KEY in your shell to copy-and-run them.

from pictograph import Client
client = Client()  # reads PICTOGRAPH_API_KEY

The client.workflows resource is the graph-workflows feature. It is separate from the pictograph.pipelines module (full_pipeline and friends), which chains upload, annotate, and train steps for convenience. (That orchestrator module was named pictograph.workflows before SDK 1.7.0; the old import path still works as a deprecated alias.)

The graph

A graph is {"version": 1, "nodes": [...], "edges": [...]}. Each node is {"id", "type", "config"}; each edge is {"source", "sourceHandle", "target", "targetHandle"}. Block types and what they consume / produce:

Block	In	Out	Config highlights
source	—	frames	kind (image/video/dataset), gcs_uri or dataset_project_id, sample_fps
model	frames	detections	model_id (a ready model), confidence_threshold, class_filter
merge	detections	detections	strategy (consensus/union/intersection), min_votes, iou_threshold
filter	detections	detections	keep_classes, min_confidence
track	detections	detections	tracker (bytetrack/botsort)
step	detections	events	step_type (line_cross/dwell/occupancy), geometry
visualize	det + events	frames	draw toggles
sink	det/events/frames	—	kind (json/csv/webhook/annotated_video)

The structural rules are validated server-side before a run starts: one source, no loops, every block connected, and a counting/dwell step must sit downstream of a track, which must sit downstream of a model. A single-image source cannot use track or step (no temporal axis).

Ensembles — running several models together

Two detectors rarely fail the same way. A workflow can run up to three models over the same frames and combine them with a merge block:

strategy	Keeps	Use it for
consensus	detections at least min_votes models found (default 2)	fewer false positives, balanced — the default
union	everything any model found	maximum recall
intersection	only what every model found	maximum precision

Two models are “the same detection” when they share a class name and their boxes overlap by at least iou_threshold (default 0.5). A model votes at most once per object, so a model that fires twice on one thing cannot manufacture its own consensus. The surviving box is the highest-confidence one, unchanged — coordinates are never averaged into a box no model predicted. Its confidence becomes the mean across the models that voted, and each detection carries votes and model_count.

{
  "nodes": [
    {"id": "s",  "type": "source", "config": {"kind": "video", "gcs_uri": "gs://...", "sample_fps": 5}},
    {"id": "m1", "type": "model",  "config": {"model_id": "<uuid-a>"}},
    {"id": "m2", "type": "model",  "config": {"model_id": "<uuid-b>"}},
    {"id": "mg", "type": "merge",  "config": {"strategy": "consensus", "min_votes": 2, "iou_threshold": 0.5}},
    {"id": "out","type": "sink",   "config": {"kind": "json"}}
  ],
  "edges": [
    {"source": "s",  "target": "m1"}, {"source": "s",  "target": "m2"},
    {"source": "m1", "target": "mg"}, {"source": "m2", "target": "mg"},
    {"source": "mg", "target": "out"}
  ]
}

More than one model block is only valid when all of them feed the merge block — a model wired around it would be run and billed and then ignored, so the run is rejected (models_need_merge / model_not_merged) rather than quietly producing the wrong answer.

An ensemble costs about as many times as it has models: every model runs over every frame, and you are billed for the GPU time that takes.

create

Create a workflow from a graph. Returns a Workflow in draft status. Management calls (create, update, delete, run, cancel) require member+ role.

graph = {
    "version": 1,
    "nodes": [
        {"id": "s", "type": "source", "config": {"kind": "video", "gcs_uri": "gs://.../clip.mp4", "sample_fps": 5}},
        {"id": "m", "type": "model", "config": {"model_id": "your-model-uuid", "confidence_threshold": 0.4}},
        {"id": "f", "type": "filter", "config": {"keep_classes": ["car", "truck"]}},
        {"id": "k", "type": "sink", "config": {"kind": "json"}},
    ],
    "edges": [
        {"source": "s", "target": "m"},
        {"source": "m", "target": "f"},
        {"source": "f", "target": "k"},
    ],
}

wf = client.workflows.create("Vehicle detector", graph)
print(wf.id, wf.status)   # 'draft'

curl -s -X POST "https://api.pictograph.io/api/v1/developer/workflows/" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Vehicle detector",
    "graph": {
      "version": 1,
      "nodes": [
        {"id": "s", "type": "source", "config": {"kind": "video", "gcs_uri": "gs://.../clip.mp4", "sample_fps": 5}},
        {"id": "m", "type": "model", "config": {"model_id": "your-model-uuid", "confidence_threshold": 0.4}},
        {"id": "f", "type": "filter", "config": {"keep_classes": ["car", "truck"]}},
        {"id": "k", "type": "sink", "config": {"kind": "json"}}
      ],
      "edges": [
        {"source": "s", "target": "m"},
        {"source": "m", "target": "f"},
        {"source": "f", "target": "k"}
      ]
    }
  }'

Returns a Workflow.

list

List every workflow in your organization.

for w in client.workflows.list():
    print(w.id, w.name, w.status)

curl -s "https://api.pictograph.io/api/v1/developer/workflows/" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns Workflow objects.

get

Fetch a single workflow by UUID. The response includes a validation array (empty when the graph is runnable).

wf = client.workflows.get(wf.id)
print(wf.status, wf.validation)

curl -s "https://api.pictograph.io/api/v1/developer/workflows/{workflow_id}" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns a Workflow.

update

Update a workflow’s name, description, graph, or status. Valid statuses are draft, ready, and archived. The response includes the refreshed validation array.

wf = client.workflows.update(wf.id, status="ready")

curl -s -X PATCH "https://api.pictograph.io/api/v1/developer/workflows/{workflow_id}" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"status": "ready"}'

Returns a Workflow.

delete

Delete a workflow and its run history.

client.workflows.delete(wf.id)

curl -s -X DELETE "https://api.pictograph.io/api/v1/developer/workflows/{workflow_id}" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

run

Validate the graph and start a run. The SDK raises ValidationError / 400 with the specific issues if the graph is not runnable, and PaymentRequiredError / 402 if there is not enough compute credit. It returns a WorkflowRunCreated with the run_id.

created = client.workflows.run(wf.id)
print(created.run_id, created.deposit_micro_usd)

curl -s -X POST "https://api.pictograph.io/api/v1/developer/workflows/{workflow_id}/run" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

get_run

Poll the run until it finishes, then read its artifacts (signed download URLs):

import time

while True:
    run = client.workflows.get_run(created.run_id)
    print(run.status, f"{run.frames_done}/{run.frames_total}", f"{run.progress:.0f}%")
    if run.status in ("completed", "error", "cancelled"):
        break
    time.sleep(3)

if run.status == "completed":
    for art in run.artifacts:
        print(art["kind"], art.get("download_url"))
    # run.step_results carries the per-step aggregates + time series

# Poll this endpoint until status is completed / error / cancelled.
curl -s "https://api.pictograph.io/api/v1/developer/workflows/runs/{run_id}" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns a WorkflowRun.

Field	Meaning
status	queued, processing, completed, error, or cancelled.
progress	0–100.
frames_done / frames_total	Frames processed / estimated.
step_results	Per step block: aggregate (headline number), time series, per-track events.
artifacts	[{kind, gcs_path, bytes, download_url}] for JSON / CSV / annotated-video outputs.
error	Set when status == "error".

cancel_run

Stop an in-flight run. This stops the GPU job and refunds the deposit.

client.workflows.cancel_run(created.run_id)

curl -s -X POST "https://api.pictograph.io/api/v1/developer/workflows/runs/{run_id}/cancel" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

bulk_cancel_runs

Cancel many runs in one call. Each run gets a Modal cancel plus an idempotent deposit refund. Foreign, terminal, or missing ids land in not_found rather than failing the call.

result = client.workflows.bulk_cancel_runs(["run-1", "run-2"])
print(result.count, "cancelled;", result.not_found, "skipped")

curl -s -X POST "https://api.pictograph.io/api/v1/developer/workflows/runs/bulk-cancel" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"run_ids": ["run-1", "run-2"]}'

Returns BulkActionResult with succeeded, not_found, and count.