# How to Build Declarative Pipelines with PipelineBuilder and set_parallelism() in Semantica

> Learn to build declarative pipelines in Semantica using PipelineBuilder and set_parallelism() to control concurrent worker execution for efficient processing. Optimize your workflows today.

- Repository: [Semantica /semantica](https://github.com/semantica-agi/semantica)
- Tags: how-to-guide
- Published: 2026-08-17

---

**Use `PipelineBuilder` to construct a directed-acyclic graph of processing steps, then call `set_parallelism(level)` before `build()` to specify the maximum number of concurrent workers for the `ExecutionEngine`.**

The semantica-agi/semantica repository provides a high-level DSL for constructing data-processing workflows through declarative pipelines. By leveraging the **PipelineBuilder** class, you can define complex DAGs of computation steps while controlling execution concurrency through the `set_parallelism()` method.

## Understanding the PipelineBuilder Architecture

The declarative pipeline system separates graph definition from execution through four core components defined in the source code:

- **PipelineBuilder** ([`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py)): A mutable builder that collects step definitions, dependencies, and execution options. It produces an immutable `Pipeline` object via the `build()` method.

- **Pipeline** ([`semantica/pipeline/pipeline.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline.py)): The immutable representation of the DAG that stores step metadata, parallelism settings, and configuration required by the execution engine.

- **ExecutionEngine** ([`semantica/pipeline/execution_engine.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/execution_engine.py)): Traverses the `Pipeline` DAG and schedules steps according to the configured parallelism level, propagating success or failure status through the graph.

- **PipelineValidator** ([`semantica/pipeline/pipeline_validator.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_validator.py)): Validates logical consistency before runtime, checking for cycles, missing inputs, and parallelism constraints.

## Configuring Parallelism with set_parallelism()

Parallelism in Semantica pipelines is controlled through the `set_parallelism()` method defined in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) around line 175:

```python
def set_parallelism(self, level: int) -> "PipelineBuilder":
    """Set the maximum number of concurrent steps."""
    # implementation …

    return self

```

When you invoke `set_parallelism(level)`, you specify how many worker threads or async slots the **ExecutionEngine** may use when scheduling independent steps. The default value is `1`, which enforces pure sequential execution. Raising this level allows the engine to execute steps with no unmet dependencies simultaneously, significantly reducing wall-clock time for I/O-bound or embarrassingly parallel workloads.

The parallelism setting is stored inside the resulting `Pipeline` object when you call `build()`, and the `ExecutionEngine.run()` method reads this value to constrain its scheduler.

## Building Declarative Pipelines (Step-by-Step Examples)

### Basic Linear Pipeline

Start by importing `PipelineBuilder` and `ExecutionEngine`, then chain step definitions using the fluent DSL:

```python
from semantica.pipeline import PipelineBuilder, ExecutionEngine

# Create a builder

builder = PipelineBuilder()

# Add steps sequentially

builder.add_step(
    name="extract_data",
    func=my_extract,
    outputs=["raw"]
).add_step(
    name="transform",
    func=my_transform,
    inputs=["raw"],
    outputs=["clean"]
).add_step(
    name="load",
    func=my_load,
    inputs=["clean"]
)

# Build and execute

pipeline = builder.build()
engine = ExecutionEngine()
engine.run(pipeline)

```

This constructs a simple linear dependency chain where each step waits for the previous one to complete.

### Enabling Concurrent Execution

To allow independent steps to run in parallel, set the parallelism level before building:

```python
builder = PipelineBuilder()

# ... define steps ...

# Allow up to 4 concurrent steps

builder.set_parallelism(level=4)

pipeline = builder.build()
ExecutionEngine().run(pipeline)

```

With `set_parallelism(4)`, the engine maintains a pool of four workers that can execute steps simultaneously when the dependency graph permits.

### Branching and Join Patterns

Real-world pipelines often ingest data from multiple independent sources before merging results. Define branches by creating steps with no mutual dependencies:

```python
builder = PipelineBuilder()

# Independent ingestion pipelines

builder.add_step("ingest_a", func=ingest_a, outputs=["a"])
builder.add_step("ingest_b", func=ingest_b, outputs=["b"])

# Join point requiring both outputs

builder.add_step(
    "merge",
    func=merge,
    inputs=["a", "b"],
    outputs=["merged"]
)

# Final processing

builder.add_step(
    "analyze",
    func=analyse,
    inputs=["merged"]
)

# Enable 3 concurrent workers

builder.set_parallelism(3)

pipeline = builder.build()
ExecutionEngine().run(pipeline)

```

In this configuration:

- `ingest_a` and `ingest_b` execute simultaneously because they share no dependencies.
- `merge` blocks until both inputs are available.
- With `set_parallelism(3)`, the engine can overlap the two ingestion steps and schedule `merge` immediately when ready.

### Validating Before Execution

Use `PipelineValidator` to catch logical errors before runtime:

```python
from semantica.pipeline import PipelineValidator

validator = PipelineValidator()
issues = validator.validate(builder)  # Pass the builder or built pipeline

if issues:
    raise ValueError(f"Pipeline validation failed: {issues}")

pipeline = builder.build()
ExecutionEngine().run(pipeline)

```

The validator checks for cycles, missing inputs, duplicate step names, and warns if the parallelism level exceeds available CPU cores on the host.

## Summary

- **PipelineBuilder** in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) provides a fluent DSL for defining DAGs of computation steps.
- Call `set_parallelism(level)` before `build()` to control concurrency; the default is `1` (sequential execution).
- The `ExecutionEngine` reads the parallelism setting from the immutable `Pipeline` object to manage its worker pool.
- Use `PipelineValidator` to verify graph integrity and parallelism constraints before execution.
- Independent steps with no mutual dependencies run concurrently only when parallelism is greater than `1`.

## Frequently Asked Questions

### What is the default parallelism level in PipelineBuilder?

The default parallelism level is `1`, which forces the `ExecutionEngine` to run steps sequentially even when the dependency graph would allow concurrent execution. You must explicitly call `set_parallelism()` with a value greater than `1` to enable parallel processing of independent steps.

### Where is the set_parallelism() method defined in the Semantica source code?

The `set_parallelism()` method is implemented in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) around line 175. It accepts an integer `level` parameter and returns the `PipelineBuilder` instance to support method chaining in the declarative DSL.

### Can I change the parallelism level after calling build()?

No. The `Pipeline` object created by `build()` is immutable. Once constructed, the parallelism level is fixed inside the pipeline configuration. To use a different concurrency level, you must create a new builder or reconfigure the existing one before calling `build()`.

### How does the ExecutionEngine handle steps that finish at different rates?

The `ExecutionEngine` maintains a pool of workers sized according to the `set_parallelism()` value. As steps complete and free up workers, the engine traverses the DAG to identify ready steps—those with all dependencies satisfied—and schedules them immediately on available workers without waiting for slower branches to finish.