# How the PipelineBuilder DSL Handles Parallel Execution in Semantica

> Learn how Semantica's PipelineBuilder DSL manages parallel execution. Discover how it uses set_parallelism and ParallelismManager to run independent steps concurrently via thread or process pools.

- Repository: [Semantica /semantica](https://github.com/semantica-agi/semantica)
- Tags: internals
- Published: 2026-08-09

---

**The `PipelineBuilder` DSL enables declarative parallel execution by storing a worker count via `set_parallelism()`, embedding that value into pipeline metadata during `build()`, and delegating step scheduling to `ParallelismManager` which groups independent steps and executes them via thread or process pools.**

The `semantica-agi/semantica` repository provides a declarative domain-specific language for constructing AI workflows that automatically optimizes throughput through parallel execution. By configuring parallelism at the DSL level, developers can scale pipeline workloads across multiple workers without manually managing concurrency primitives or dependency graphs.

## Declaring Parallelism in the DSL

The `PipelineBuilder` class exposes two methods for declaring concurrency levels depending on your construction style.

**Method-based configuration** uses `set_parallelism(level: int)` defined in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) (lines 175-186). This stores the desired worker count in the builder’s internal `pipeline_config` dictionary:

```python
builder = PipelineBuilder()
builder.set_parallelism(4)  # Enable 4 parallel workers

```

**Dictionary-based configuration** accepts a `parallelism` key directly in the configuration payload passed to `build_pipeline()`. When provided, this value is extracted and stored in the same `pipeline_config` structure.

## Embedding Configuration in Pipeline Metadata

When `build()` or `build_pipeline()` constructs the final `Pipeline` instance, the stored parallelism value is retrieved from `pipeline_config` and attached to the pipeline’s metadata. In [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) (lines 221-224), the implementation embeds the configuration as follows:

```python
"parallelism": self.pipeline_config.get("parallelism", 1)

```

This ensures every pipeline carries its concurrency requirement as metadata, defaulting to single-threaded execution (level `1`) when not explicitly specified.

## Runtime Parallel Execution

The `ExecutionEngine` orchestrates pipeline runs by delegating step scheduling to the `ParallelismManager`. This separation of concerns allows the DSL to remain purely declarative while the execution layer handles the complexity of concurrent task management.

### Identifying Parallelizable Step Groups

Before execution begins, the `ParallelismManager` analyzes the pipeline’s dependency graph to identify steps that can run concurrently. The method `identify_parallelizable_steps` (lines 321-332 in [`semantica/pipeline/parallelism_manager.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/parallelism_manager.py)) groups steps that share no inter-dependencies, ensuring that dependent steps execute sequentially while independent branches run simultaneously.

### Thread and Process Pool Execution

Once step groups are identified, the manager dispatches them via `execute_parallel` (lines 20-48 in [`semantica/pipeline/parallelism_manager.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/parallelism_manager.py)). By default, the system uses a `ThreadPoolExecutor` for concurrent execution. For CPU-bound workloads, you can enable process-based parallelism by setting `use_processes=True`, which triggers the `_execute_with_threads` alternative path (lines 71-77).

The core execution flow sorts tasks by priority, dispatches them to the pool, aggregates results, and updates each `PipelineStep`’s status to `StepStatus.COMPLETED` or `FAILED` upon completion.

## Practical Implementation Examples

### Basic DSL Usage with Method Chaining

Configure parallelism declaratively using the builder’s fluent interface:

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

# Build a pipeline with 4 parallel workers

builder = PipelineBuilder()
pipeline = (
    builder.add_step("ingest", "file_ingest", source="./docs")
           .add_step("parse", "document_parse", dependencies=["ingest"])
           .add_step("embed", "text_embed", dependencies=["parse"])
           .set_parallelism(4)          # ← enable parallelism

           .build(name="my_parallel_pipeline")
)

engine = ExecutionEngine()
result = engine.execute_pipeline(pipeline)
print(result.success, result.metrics["steps_executed"])

```

### Configuration-Based Setup

Define parallelism within a dictionary payload for dynamic configuration:

```python
from semantica.pipeline import PipelineBuilder

config = {
    "name": "cfg_parallel",
    "steps": [
        {"name": "step_a", "type": "ingest", "config": {"source": "./data"}},
        {"name": "step_b", "type": "parse", "dependencies": ["step_a"]},
        {"name": "step_c", "type": "embed", "dependencies": ["step_b"]},
    ],
    "parallelism": 3,                     # ← parallel workers defined here

}

builder = PipelineBuilder()
pipeline = builder.build_pipeline(config)

```

### Advanced Manual Execution

For fine-grained control, interact directly with `ParallelismManager` to execute arbitrary tasks:

```python
from semantica.pipeline import ParallelismManager, Task

def fetch(url):
    # …fetch logic…

    return url

tasks = [
    Task(task_id="t1", handler=fetch, args=("https://a.com",)),
    Task(task_id="t2", handler=fetch, args=("https://b.com",)),
    Task(task_id="t3", handler=fetch, args=("https://c.com",)),
]

manager = ParallelismManager(max_workers=3)
results = manager.execute_parallel(tasks)
for r in results:
    print(r.task_id, r.success, r.result)

```

## Summary

- **Declarative Configuration**: Use `set_parallelism(level)` or the `parallelism` configuration key to specify worker counts without managing threads manually.
- **Metadata Attachment**: The `PipelineBuilder.build()` method embeds parallelism settings into pipeline metadata at lines 221-224 of [`pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/pipeline_builder.py).
- **Automatic Grouping**: `ParallelismManager.identify_parallelizable_steps` (lines 321-332) automatically detects independent steps that can run concurrently.
- **Flexible Execution**: The system defaults to `ThreadPoolExecutor` via `execute_parallel` (lines 20-48) but supports process pools via `use_processes=True`.
- **Dependency Respect**: Parallel execution preserves explicit step dependencies while maximizing throughput for independent branches.

## Frequently Asked Questions

### How do I enable parallel execution in a Semantica pipeline?

Call `set_parallelism(level: int)` on your `PipelineBuilder` instance with the desired number of workers, or include a `"parallelism": N` key in your configuration dictionary. According to the source code in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py), this value is stored in `pipeline_config` and embedded into the pipeline metadata during the build phase.

### What is the default parallelism level if not specified?

The system defaults to single-threaded execution. In [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) (lines 221-224), the build method uses `self.pipeline_config.get("parallelism", 1)`, which returns `1` when no parallelism level has been declared.

### Does Semantica use threads or processes for parallel execution?

By default, `ParallelismManager` uses a `ThreadPoolExecutor` for concurrent step execution, as implemented in `execute_parallel` (lines 20-48 of [`parallelism_manager.py`](https://github.com/semantica-agi/semantica/blob/main/parallelism_manager.py)). However, you can enable process-based parallelism by setting `use_processes=True`, which utilizes the process pool implementation found at lines 71-77.

### How does the system determine which steps can run in parallel?

The `ParallelismManager` analyzes the pipeline’s dependency graph using `identify_parallelizable_steps` (lines 321-332 in [`semantica/pipeline/parallelism_manager.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/parallelism_manager.py)). Steps that do not depend on each other’s outputs are grouped together and executed concurrently, while maintaining sequential execution for dependent steps.