Skip to content

Workflow Engine

The workflow execution engine provides powerful workflow parsing, validation, and execution with matrix expansion and template variable support.

Package Structure

The engine is a package (causaliq_workflow.workflow) of focused mixin modules composed into WorkflowExecutor. The package re-exports the public names so existing imports keep resolving:

  • matrix — matrix expansion, derivation and value matching;
  • parameters — workflow parameter names, binding and validation;
  • templates — template variable extraction, resolution and step-name expansion;
  • patterns — step pattern detection;
  • filters — shared reporting view construction and filter resolution;
  • aggregation — AGGREGATE pattern configuration and input scanning;
  • validation / validation_entries — configuration and per-entry validation;
  • execution — parsing and job execution;
  • subworkflow / frames — calling another workflow;
  • provenance — provenance collection for cache entries;
  • executor — the composed WorkflowExecutor.

Core Classes

causaliq_workflow.workflow

WorkflowExecutor 

WorkflowExecutor()

Parse and execute GitHub Actions-style workflows.

This class handles the parsing of YAML workflow files and expansion of matrix strategies into individual experiment jobs. It provides the foundation for executing multi-step causal discovery workflows with parameterised experiments using flexible action parameter templating.

Methods:

parse_workflow 

parse_workflow(
    workflow_path: Union[str, Path],
    mode: str = "dry-run",
    parameters: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]

Parse workflow YAML file with validation.

Parameters:

  • workflow_path
    (Union[str, Path]) –

    Path to workflow YAML file

  • mode
    (str, default: 'dry-run' ) –

    Execution mode for action validation

  • parameters
    (Optional[Dict[str, Any]], default: None ) –

    Parameter names supplied by the CLI, so a workflow may reference them in its templates before execution

Returns:

  • Dict[str, Any] –

    Parsed and validated workflow dictionary

Raises:

expand_matrix 

expand_matrix(matrix: Dict[str, List[Any]]) -> List[Dict[str, Any]]

Expand matrix variables into individual job configurations.

Generates all combinations from matrix variables using cartesian product. Each combination becomes a separate job configuration.

Range strings like "0-24" are automatically expanded to individual integer values [0, 1, 2, ..., 24].

Parameters:

  • matrix
    (Dict[str, List[Any]]) –

    Dictionary mapping variable names to lists of values

Returns:

  • List[Dict[str, Any]] –

    List of job configurations with matrix variables expanded

Raises:

execute_workflow 

execute_workflow(
    workflow: Dict[str, Any],
    mode: str = "dry-run",
    cli_params: Optional[Dict[str, Any]] = None,
    step_logger: Optional[StepLogger] = None,
    validate: bool = True,
    source_path: Optional[Union[str, Path]] = None,
) -> List[Dict[str, Any]]

Execute complete workflow with matrix expansion.

Uses two-pass execution:

  1. Validation pass: Iterates all entries (matrix combos/cache entries) and validates action parameters. This catches semantic errors like undefined filter variables before any execution.
  2. Execution pass: If validation passes, executes the workflow.

Caching is controlled at the step level via the 'output' parameter in each step's 'with' block. Each step can write to its own cache.

Parameters:

  • workflow
    (Dict[str, Any]) –

    Parsed workflow dictionary

  • mode
    (str, default: 'dry-run' ) –

    Execution mode ('dry-run', 'run', 'compare')

  • cli_params
    (Optional[Dict[str, Any]], default: None ) –

    Additional parameters from CLI

  • step_logger
    (Optional[StepLogger], default: None ) –

    Optional function to log step execution (action_method, step_name, status, matrix_values). The optional depth and detail keyword arguments carry the workflow nesting depth and a trailing detail fragment.

  • validate
    (bool, default: True ) –

    When True (default) run the validation pass first. Set False when the caller has already validated the same workflow to avoid a redundant pass.

  • source_path
    (Optional[Union[str, Path]], default: None ) –

    Path of the workflow file used to resolve relative calls to other workflows. Defaults to the last workflow parsed by this executor.

Returns:

  • List[Dict[str, Any]] –

    List of job results from matrix expansion

Raises:

Exception Handling

WorkflowExecutionError

Raised when workflow execution fails.

Usage Examples

Basic Workflow Execution

from causaliq_workflow import WorkflowExecutor, WorkflowExecutionError
from causaliq_workflow.registry import ActionRegistry

# Create executor instance
executor = WorkflowExecutor()

try:
    # Parse and validate workflow (includes template variable validation)
    workflow = executor.parse_workflow("experiment.yml")
    # Keys other than matrix and steps are workflow parameters
    print(f"Workflow ID: {workflow.get('id')}")
    print(f"Sample size: {workflow.get('sample_size')}")

    # Execute the complete workflow
    results = executor.execute_workflow(workflow, mode="run")
    print(f"Workflow completed successfully")

except WorkflowExecutionError as e:
    print(f"Workflow execution failed: {e}")

Workflow Parameters

Declared parameters are overridden with parameters when parsing (so the parser knows the names a workflow expects) and with cli_params when executing:

workflow = executor.parse_workflow(
    "experiment.yml", parameters={"sample_size": 5000}
)
results = executor.execute_workflow(
    workflow, mode="run", cli_params={"sample_size": 5000}
)

A name that matches a matrix dimension replaces that dimension; other names override a declared default or fill a {{name}} placeholder. A call step's with block uses the same rules; see Reusable Workflows.

Workflow Calls

A step can call another workflow by using a relative uses path. parse_workflow remembers the workflow's location, so relative calls resolve correctly:

workflow = executor.parse_workflow("workflows/parent.yml")
results = executor.execute_workflow(workflow, mode="run")

Pass source_path when executing a workflow the executor did not parse:

from causaliq_workflow.schema import load_workflow_file

workflow = load_workflow_file("workflows/parent.yml")
results = executor.execute_workflow(
    workflow, mode="run", source_path="workflows/parent.yml"
)

See Reusable Workflows for the syntax.

Matrix Expansion

from causaliq_workflow import WorkflowExecutor

executor = WorkflowExecutor()

# Define matrix for parameter sweeps
matrix = {
    "algorithm": ["pc", "ges", "lingam"],
    "dataset": ["asia", "cancer"],
    "alpha": [0.01, 0.05]
}

# Expand into individual parameter combinations
jobs = executor.expand_matrix(matrix)
print(f"Generated {len(jobs)} jobs from matrix")  # Results in 12 jobs (3 × 2 × 2)

for i, job in enumerate(jobs):
    print(f"Job {i}: Algorithm={job['algorithm']}, Dataset={job['dataset']}, Alpha={job['alpha']}")

Template Variable System

# The WorkflowExecutor automatically validates template variables during parsing
# Template variables ({{variable}}) are checked against available context

# Example: Valid template usage
valid_workflow = {
    "id": "test-001",
    "matrix": {"dataset": ["asia"], "algorithm": ["pc"]},
    "steps": [{
        "uses": "my-custom-action",
        "with": {
            "output": "/results/{{id}}/{{dataset}}_{{algorithm}}.xml",
            "description": "Processing {{dataset}} with {{algorithm}}"
        }
    }]
}

try:
    workflow = executor.parse_workflow("workflow.yml")
    print("Template validation passed!")
except WorkflowExecutionError as e:
    if "Unknown template variables" in str(e):
        print(f"Template validation failed: {e}")
        # Example: "Unknown template variables: ['missing_var'].
        # Available variables: ['algorithm', 'dataset', 'id']"
    else:
        print(f"Workflow execution failed: {e}")

Advanced Workflow Features

# Example workflow YAML showing flexible action parameters
workflow_yaml = \"\"\"
id: "experiment-001"
matrix:
  dataset: ["asia", "cancer"]
  algorithm: ["pc", "ges"]
  alpha: [0.01, 0.05]

steps:
  - name: "Structure Learning"
    uses: "my-structure-learner"
    with:
      data_path: "/experiments/data/{{dataset}}.csv"
      output_dir: "/experiments/results/{{id}}/{{algorithm}}/"
      alpha: {{alpha}}
      max_iter: 1000

  - name: "Validation"
    uses: "validate-graph"
    with:
      graph_path: "/experiments/results/{{id}}/{{algorithm}}/graph.graphml"
      metrics_output: "/experiments/results/{{id}}/{{algorithm}}/metrics.json"
\"\"\"

# Save and parse the workflow
with open("experiment.yml", "w") as f:
    f.write(workflow_yaml)

workflow = executor.parse_workflow("experiment.yml")

# Matrix expansion creates jobs with substituted variables
if "matrix" in workflow:
    jobs = executor.expand_matrix(workflow["matrix"])
    # Creates 8 jobs (2×2×2) with customizable file paths

    for job in jobs:
        print(f"Job: {job}")
        # Example output: {'dataset': 'asia', 'algorithm': 'pc', 'alpha': 0.01}

Template Variable Context

Template variables can reference:

  1. Workflow parameters: declared top-level keys such as {{id}}
  2. Matrix variables: any variables defined in the matrix section
  3. Supplied parameters: values from a caller's with: or --param

Declared parameters have defaults and may be overridden:

sample_size: 1000     # default 1000
reference: null       # no value: the parameter is omitted from the action

A declaration with a null default supplies no value. Unless a caller or --param supplies one, the parameter is omitted from every action that would otherwise receive it, so the action's own default applies.

Three rules apply to declarations:

  • A declared parameter must be used as a {{placeholder}} in at least one step's with block; an unused declaration is a validation error.
  • A null-default parameter may only be the whole value of a step parameter. "data/{{reference}}.csv" would substitute None and is rejected.
  • A placeholder that is not declared is effectively required, so a caller or --param must supply it.

Available Template Variables

Context Variables Example
Parameters Declared top-level keys {{id}}, {{output_root}}
Matrix User-defined matrix vars {{dataset}}, {{algorithm}}
AGGREGATE cache Keys derived from input caches {{network}}

A whole-value token may be written without quotes: sample_size: {{sample_size}} is valid. A value that is exactly one token keeps its native type. An action parameter declared as int, float or bool also accepts its text spelling, because the value is converted just before the action runs.

Unknown variables are rejected when the workflow is parsed, unless a caller or the CLI supplies them:

Unknown template variables: ['sampl_size'].
Available variables: ['algorithm', 'dataset']

The same error reports a required parameter whose value was not supplied, because an undeclared placeholder has no default.

Parameter Usage Requirement

A declared parameter must be used as a {{name}} placeholder in at least one step's with block. A declaration that no step references is reported as a validation error, because it cannot affect any action:

# Invalid: 'sample_size' is declared but never used
sample_size: 1000
matrix:
  network: [asia]
steps:
  - uses: causaliq-discovery
    with:
      action: run_pc
      network: {{network}}

A null-default parameter is also restricted to whole-value use:

reference: null
steps:
  - uses: causaliq-analysis
    with:
      action: evaluate_graph
      reference: {{reference}}             # valid: whole value
      dataset: "data/{{reference}}.csv"      # invalid: no value to embed

Matrix Variable Usage Requirement

All matrix variables must be used in at least one CREATE pattern step's parameter templates. This ensures each matrix combination produces a distinct action invocation and prevents accidental matrix variable definitions that have no effect.

# Valid: Both matrix variables are used
matrix:
  network: [asia, alarm]
  seed: [1, 2, 3]
steps:
  - uses: causaliq-discovery
    with:
      action: run_pc
      network: {{network}}
      seed: {{seed}}
      output: results.db

# Invalid: 'seed' is defined but never used
matrix:
  network: [asia, alarm]
  seed: [1, 2, 3]  # Error: unused matrix variable
steps:
  - uses: causaliq-discovery
    with:
      action: run_pc
      network: {{network}}
      output: results.db

Exceptions:

  • AGGREGATE pattern steps use matrix for grouping entries, not templating. Matrix variables don't need to appear in AGGREGATE step parameters.

Aggregation Processing

Aggregation mode enables workflow steps to consume entries from multiple cache entries. This is essential for research workflows that merge graphs or compute aggregate statistics across parameter sweeps.

Aggregation Configuration

AggregationConfig dataclass 

AggregationConfig(
    input_caches: List[str] = list(),
    filter_expr: Optional[str] = None,
    matrix_vars: List[str] = list(),
)

Configuration for aggregation mode execution.

Aggregation mode is activated when a workflow step has: - An input parameter specifying workflow cache(s) - A matrix definition in the workflow

The matrix variables define the grouping dimensions for aggregation.

Attributes:

  • filter_expr (Optional[str]) –

    Optional filter expression to restrict input entries.

  • input_caches (List[str]) –

    List of input workflow cache paths.

  • matrix_vars (List[str]) –

    Matrix variables defining grouping dimensions.

filter_expr class-attribute instance-attribute 
filter_expr: Optional[str] = None

Optional filter expression to restrict input entries.

input_caches class-attribute instance-attribute 
input_caches: List[str] = field(default_factory=list)

List of input workflow cache paths.

matrix_vars class-attribute instance-attribute 
matrix_vars: List[str] = field(default_factory=list)

Matrix variables defining grouping dimensions.

Aggregation Example

# Example workflow with aggregation
workflow_yaml = \"\"\"
id: "merge_experiment"
matrix:
  model: ["asia", "cancer"]

steps:
  - name: "Merge Graphs"
    uses: "causaliq-analysis"
    with:
      action: "merge_graphs"
      input: "results/{{id}}/llm_graphs.db"  # .db triggers aggregation
      output: "results/{{id}}/merged.db"
\"\"\"

# Execute with aggregation mode
executor = WorkflowExecutor()
workflow = executor.parse_workflow("workflow.yml")

# For each matrix value (asia, cancer), the action receives
# all cache entries from llm_graphs.db that match that model
results = executor.execute_workflow(workflow, mode="run")

Filter Expressions

Use filters to restrict which entries are aggregated:

steps:
  - name: "Merge Filtered"
    uses: "causaliq-analysis"
    with:
      action: "merge_graphs"
      input: "results/graphs.db"
      filter: learn_graph.algorithm == 'pc' and sample_size >= 1000
      output: "results/merged.db"

Error Handling

The WorkflowExecutor provides detailed error reporting for:

  • Parse errors: Invalid YAML/JSON syntax
  • Schema validation: Workflow structure validation
  • Template errors: Unknown or invalid template variables
  • Action errors: Action execution failures
  • Matrix errors: Invalid matrix definitions

← Previous: Registry | Back to API Overview | Next: Cache →