# Workflows
source: https://docs.chalk.ai/docs/compute/workflows

## Run Python workflows on dynamically provisioned scaling groups with separate images for each task.

### Overview

A workflow is an async Python function that calls tasks and combines their
results. Use workflows for jobs with multiple steps, such as preparing data,
running inference, and writing results. The workflow orchestrator records
progress and applies each task's retry and timeout settings.

Deploy a workflow with chalkcompute.deploy_workflow_worker. Chalk packages
the source file and its dependencies into an image and creates a
scaling group to run it. Each task can declare
its own image and Python dependencies. Chalk creates a separate scaling group
for each distinct task image, so steps with different environments can run in
the same workflow.

Workers start at zero replicas. Submitting a run wakes the workflow's worker;
calling a task with its own image wakes that task's worker. Each group stays
up while it has an open workflow and returns to zero after its idle delay.

### Define and deploy a workflow

Use an environment with a workflow orchestrator enabled, and authenticate
with chalk login.

Define tasks with @task and the orchestration function with @workflow.
Tasks can be synchronous or asynchronous; workflow functions must be
async def. Save this example as pipeline.py:

```
# /// script
# dependencies = ["chalkcompute", "chalkpy[workflows]"]
# ///
import chalkcompute
from chalk.workflows import task, workflow


@task
def score(rows: list[int]) -> list[float]:
    return [round(row * 0.1, 2) for row in rows]


@task
def summarize(batch_id: str, scores: list[float]) -> dict:
    return {
        "batch_id": batch_id,
        "rows": len(scores),
        "mean_score": sum(scores) / len(scores) if scores else None,
    }


@workflow
async def score_batch(batch_id: str, rows: list[int]) -> dict:
    scores = await score(rows)
    return await summarize(batch_id, scores)


if __name__ == "__main__":
    worker = chalkcompute.deploy_workflow_worker(name="wf-score-batch")
    print("task queue:", worker.task_queue)
```

Run the script to deploy the worker:

```
uv run pipeline.py
```

The worker serves score_batch on the queue
chalkcompute-workflows.wf-score-batch. Deploying the same name again updates
the existing group. Deployment does not start a workflow run.

Keep deployment and submission code inside if __name__ == "__main__":.
Workers import the file to register its tasks and workflows, so code outside
that block also runs on each worker.

Workflow code must be deterministic. Put network requests, file access,
randomness, and wall-clock reads inside tasks. Use JSON-serializable workflow
inputs and return values.

### Submit a run

Once the worker is deployed, submit runs by name from another script or
service. The caller needs Chalk credentials and chalkpy[workflows]; it does
not need to import pipeline.py.

Save this as submit.py, using the task queue printed by the deployment:

```
# /// script
# dependencies = ["chalkpy[workflows]"]
# ///
from chalk.client import ChalkClient

run = ChalkClient().trigger_workflow(
    "score_batch",
    {"batch_id": "batch-001", "rows": list(range(8))},
    task_queue="chalkcompute-workflows.wf-score-batch",
    wait=False,
)
print("workflow id:", run.workflow_id)
print("run id:", run.run_id)
```

```
uv run submit.py
```

wait=False returns a handle once the run is accepted. The workflow continues
after the submitting process exits. Pass wait=True to wait for completion
and receive the workflow's return value instead of a handle:

```
result = ChalkClient().trigger_workflow(
    "score_batch",
    {"batch_id": "batch-001", "rows": list(range(8))},
    task_queue="chalkcompute-workflows.wf-score-batch",
    wait=True,
)
print(result)
# {'batch_id': 'batch-001', 'rows': 8, 'mean_score': 0.35}
```

The first task may wait for a worker to start and pull its image when the
group is at zero replicas.

If you already have the worker returned by deploy_workflow_worker, use
worker.trigger(...) to submit to its queue:

```
run = worker.trigger(
    score_batch,
    {"batch_id": "batch-002", "rows": [2, 4, 6]},
)
```

worker.trigger also accepts a workflow name and defaults to wait=False.
Pass wait=True to return the result.

### Task images and dependencies

Set image on a task to give it a separate Python environment. For example,
replace score in pipeline.py with a task that uses NumPy:

```
numpy_image = chalkcompute.Image.debian_slim("3.12").pip_install(["numpy"])


@task(image=numpy_image, retries=2, timeout="5m")
def score(rows: list[int]) -> list[float]:
    import numpy as np

    return np.round(np.asarray(rows) * 0.1, 2).tolist()
```

Run uv run pipeline.py again to deploy the change. Chalk creates an
additional scaling group for the NumPy image. score runs there, while
score_batch and summarize run on the main worker. Tasks with equal image
definitions share a group, including when the images are separate Python
objects. Tasks without an image use the main worker's environment.

Keep imports for task-specific packages inside the task body. Every worker
imports the whole source file; importing NumPy at module scope would require
it on the main worker and the deploying machine too.

The default worker image installs the dependencies in the script's
# /// script block, including chalkpy[workflows]. Each task image also
receives those shared dependencies and the source file. Put packages shared
by all workers in that block, and packages used only by a task on its image.
See Images for installing system packages and adding
local files.

Inspect the groups through worker.scaling_group for the main worker and
worker.task_workers for the mapping from task names to their groups.
Groups remain deployed at zero replicas between runs; scaling down does not
delete them.

### Worker configuration

Pass worker settings to deploy_workflow_worker:

```
worker = chalkcompute.deploy_workflow_worker(
    source="pipeline.py",
    name="wf-score-batch",
    cpu="2",
    memory="4Gi",
    max_replicas=2,
    shutdown_delay_seconds=60,
)
```

source defaults to the running script. Without an explicit name, the
worker name is derived from the file name, such as wf-pipeline.

The main worker and task image workers use the same CPU, memory, replica
limit, shutdown delay, env, and secrets settings. The helper sets
min_replicas=0 and configures each group to scale on its workflow task
queue. An open workflow keeps its group running, including while it waits
for a task or retry.

Pass image to customize the main worker's image. With a custom main image,
install the script's dependencies and chalkpy[workflows] yourself; the
helper adds the source file.

### Retries and parallel tasks

Configure retries and timeouts on each @task. In the NumPy example,
retries=2 permits two retries after the first failed attempt, and
timeout="5m" limits each attempt to five minutes. The defaults are no
retries and a ten-minute timeout per attempt. Use retry_delay, such as
retry_delay="30s", to set the initial retry interval.

A retried task can execute more than once. Make external writes safe to
repeat, for example by using a batch ID as an idempotency key.

Use .submit() inside a workflow to start independent tasks before waiting
for their results:

```
import asyncio


@workflow
async def score_batches(batches: list[list[int]]) -> list[list[float]]:
    handles = [score.submit(rows) for rows in batches]
    return await asyncio.gather(*handles)
```

Each call uses the task's image, retry policy, and timeout. Add the workflow
to pipeline.py and redeploy the worker before submitting it by name.

### Run locally

Call a workflow directly to test it in the current process:

```
import asyncio
from pipeline import score_batch

result = asyncio.run(score_batch("local-batch", [2, 4, 6]))
```

Direct calls use the in-process orchestrator. They do not deploy scaling
groups or use task images, so install each task's dependencies locally.

To execute locally with durable coordination from the environment's
orchestrator, use ChalkClient().run_workflow:

```
from chalk.client import ChalkClient

result = ChalkClient().run_workflow(
    score_batch,
    {"batch_id": "local-batch", "rows": [2, 4, 6]},
)
```

This call blocks until the workflow returns. Its tasks execute in the
calling process and ignore image; keep that process running for the
duration of the workflow.