Skip to contents

Run dsprrr in pipelines

Source: vignettes/orchestration.Rmd
orchestration.Rmd

This page is about running dsprrr outside an interactive session. It shows how to start a targets pipeline and a Quarto report from dsprrr’s templates, check inputs with validate_workflow(), keep evaluation results in a pins board, and control concurrency and cost in batch runs. Saving and restoring the module itself is covered in Tutorial 6: Save and reuse a module.

Start from the templates

use_dsprrr_template() copies a targets pipeline ("targets"), a Quarto report ("quarto") or both ("all") into a folder. Existing files are kept unless overwrite = TRUE.

use_dsprrr_template("all", path = "sentiment-pipeline")
#> Created: sentiment-pipeline/_targets.R
#> Created: sentiment-pipeline/report.qmd

_targets.R is a complete sentiment-classification pipeline. Replace the example data and module with your own; the other targets wire them together:

Target What it does
train_data, test_data Six training and three test reviews with a target column
module_definition A sentiment module with instructions and a template
llm_client The Chat, rebuilt on every run (see below)
optimized_module A copy of the module tuned by optimize_grid() over temperature
evaluation_results evaluate() on the test data
pins_board, pinned_config, pinned_evaluation A versioned pins::board_folder("pins") holding the optimized module and the evaluation
pinned_traces Meant to pin traces; it writes nothing for now, because the stored module has no recorded traces
summary_stats, summary_json A summary list, also written to outputs/summary.json

llm_client calls Claude Sonnet 4.5 through ellmer::chat_anthropic(). Set the DSPRRR_MODEL environment variable for another Claude model. For any other Chat, set a factory near the top of _targets.R; it receives the model name:

options(dsprrr.targets.llm_factory = function(model) {
  ellmer::chat_openai(model = "gpt-6-luna")
})

From the pipeline’s folder, run it and read results with targets:

library(targets)
tar_make()
tar_read(evaluation_results)

On later runs targets rebuilds llm_client, sees the same Chat, and skips every target whose code and inputs haven’t changed: edit the test data and only the evaluation and what depends on it run again. Because llm_client is always rebuilt, tar_outdated() lists everything downstream of it even when tar_make() would skip those targets.

Render the report

report.qmd reads pins from the board named by its pins_board_path parameter. Its module_name, eval_name and traces_name parameters default to the pin names the pipeline writes. After tar_make(), render it with the Quarto CLI, overriding parameters with -P:

quarto render report.qmd -P eval_name:sentiment-eval-results

Treat the report as a starting point. It needs R 4.4 or later (it uses %||% without loading rlang), and its evaluation section (mean score, score distribution, predictions) is the part that works today. The module section expects an older pin layout and prints empty fields, and the trace sections stay empty while pinned_traces writes nothing. To render the report as part of the pipeline, uncomment the tar_quarto() target (from tarchetypes) at the end of _targets.R.

Check inputs before a run

validate_workflow() makes no LLM calls. It checks that module is a dsprrr module with at least one input, that data has a column for every signature input, and that board is a pins board. It prints a summary and invisibly returns a list with valid and the individual checks:

classify <- module(
  signature("text -> sentiment: enum('positive', 'negative', 'neutral')")
)
reviews <- tibble::tibble(
  text = c("Great value", "Arrived broken", "It's a chair"),
  sentiment = c("positive", "negative", "neutral")
)
board <- pins::board_folder("sentiment-pipeline/pins", versioned = TRUE)

validate_workflow(classify, data = reviews, board = board)
#> 
#> ── Workflow Validation ──
#> 
#> ✔ module: Module type: PredictModule
#> ✔ signature: 1 input(s) defined
#> ✔ data: 3 rows, 1 required columns present
#> ✔ board: Board type: pins_board_folder
#> ✔ Workflow validation passed

A missing column fails the check:

check <- validate_workflow(classify, data = tibble::tibble(review = "Great"))
#> 
#> ── Workflow Validation ──
#> 
#> ✔ module: Module type: PredictModule
#> ✔ signature: 1 input(s) defined
#> ✖ data: Missing columns: text
#> ✖ Workflow validation failed
check$valid
#> [1] FALSE

It does not look at column types, the Chat or API keys, and it does not read or write the board. Stop a script early with if (!check$valid) stop("Workflow validation failed").

Keep evaluation results

pin_vitals_log() stores an evaluate() result (mean score, per-row scores, predictions) with basic module information. Despite the name, it doesn’t accept a vitals Task; vitals writes its own logs (see Evaluate with vitals).

result <- evaluate(
  classify,
  reviews,
  metric = metric_exact_match(field = "sentiment"),
  .llm = ellmer::chat_openai(model = "gpt-6-luna")
)
pin_vitals_log(board, "sentiment-eval", result, module = classify)

pins::pin_versions(board, "sentiment-eval")
pins::pin_read(board, "sentiment-eval")$mean_score

Each write to a versioned board adds a version; pass an id from pin_versions() to pin_read(version = ) to read an older one. For shared storage, use another pins board, such as pins::board_s3("my-bucket", prefix = "dsprrr/") or pins::board_connect().

Control concurrency and cost

Batch calls run one row at a time unless you pass a concurrency_control() to .concurrency in run(), run_dataset(), evaluate() or as_vitals_task():

control <- concurrency_control(backend = "ellmer", max_active = 8L, max_errors = 5L)
control
#> <dsprrr_concurrency_control>
#> Backend: ellmer
#> Maximum active: 8
#> Task timeout: unlimited
#> Total timeout: unlimited
#> Maximum errors: 5
#> Cancel active work: TRUE
Backend How rows run Limits
"sequential" (default) One at a time Reuses cached responses
"ellmer" Concurrently in this R session, through ellmer::parallel_chat_structured() No finite timeouts; skips the response cache
"mirai" In background R processes The only backend for task_timeout and total_timeout; the Chat must come from the module or the default, not .llm; skips the response cache
"auto" Mirai when a timeout is set; otherwise sequential for max_active = 1, else ellmer when the Chat supports it, then mirai, then sequential

max_errors stops scheduling new rows after that many failures, and cancel decides whether running mirai tasks are stopped at a limit. In structured results, each row’s metadata records the backend that ran it.

results <- run(
  classify,
  text = reviews$text,
  .llm = ellmer::chat_openai(model = "gpt-6-luna"),
  .concurrency = control,
  .return_format = "structured"
)

get_cost(results)$total # dollars for this batch
summarize_traces(classify) # tokens, cost and latency recorded on this module

get_cost() reads structured run() results and evaluate() results; an unknown price stays NA rather than counting as free. session_cost() sums dsprrr’s prompt history, which keeps the last 100 calls (raise the limit with options(dsprrr.prompt_history_max = )). Calls made on the ellmer backend currently enter that history without tokens or cost, so its totals read NA after such a batch. To keep a module’s call records, pin_trace(board, name, module) pins them; add include_prompts = TRUE and include_outputs = TRUE to keep the text as well.

For the settings that apply to each Chat, such as timeouts and retries, see Models, providers and streaming.