# How the Pipeline DSL in Semantica Manages Parallel Execution and Dependencies: A Complete Guide

> Discover how Semantica's pipeline DSL masters parallel execution and dependencies. Learn to manage worker counts and automatically run independent steps concurrently for efficient workflows.

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

---

**The Semantica pipeline DSL enables parallel execution through a declarative `set_parallelism()` method that stores worker counts in pipeline metadata, while the `ExecutionEngine` automatically groups independent steps and runs them concurrently via `ParallelismManager`.**

The `PipelineBuilder` DSL in semantica-agi/semantica provides a fluent interface for constructing data processing workflows. Beyond simple sequential execution, it offers sophisticated parallel execution capabilities that respect step dependencies. This article examines how the DSL enables parallelism declaration, how the execution engine schedules concurrent work, and how dependencies constrain but don't prevent parallelization.

## Declaring Parallelism in the Pipeline DSL

The entry point for parallel execution is the `set_parallelism(level: int)` method defined in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py). This method stores the desired worker count in the builder's internal `pipeline_config` dictionary.

```python

# From semantica/pipeline/pipeline_builder.py (lines 175-186)

def set_parallelism(self, level: int) -> "PipelineBuilder":
    """
    Set the number of parallel workers for this pipeline.
    
    Args:
        level: Number of concurrent workers (default: 1)
    
    Returns:
        Self for method chaining
    """
    self.pipeline_config["parallelism"] = max(1, level)
    return self

```

Key characteristics of this declaration:

- **Fluent interface** — returns `PipelineBuilder` to enable method chaining with `.add_step()`
- **Validation** — uses `max(1, level)` to ensure at least one worker
- **Storage location** — persists in `self.pipeline_config` until `build()` is called

### Alternative Configuration-Based Declaration

Parallelism can also be declared via configuration dictionary, which `build_pipeline()` processes identically:

```python
from semantica.pipeline import PipelineBuilder

config = {
    "name": "data_processing",
    "steps": [
        {"name": "ingest", "type": "file_ingest"},
        {"name": "transform", "type": "normalize", "dependencies": ["ingest"]},
        {"name": "export", "type": "save", "dependencies": ["transform"]},
    ],
    "parallelism": 4,  # ← declarative parallelism

}

pipeline = PipelineBuilder().build_pipeline(config)

```

## Propagating Parallelism to Pipeline Metadata

When `build()` or `build_pipeline()` creates a `Pipeline` instance, the stored parallelism value attaches to pipeline metadata. This occurs in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) (lines 221-224):

```python
pipeline = Pipeline(
    name=name,
    steps=compiled_steps,
    metadata={
        "parallelism": self.pipeline_config.get("parallelism", 1),
        # ... other metadata

    },
)

```

This propagation is critical: it separates **declaration-time** configuration from **execution-time** behavior. The DSL handles the former; the execution engine handles the latter.

## How ParallelismManager Executes Steps Concurrently

The `ExecutionEngine` delegates parallel execution to `ParallelismManager` (defined in [`semantica/pipeline/parallelism_manager.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/parallelism_manager.py)). This manager performs two core operations: **dependency analysis** and **concurrent execution**.

### Step 1: Identifying Parallelizable Step Groups

Before executing any steps, the manager analyzes the dependency graph to find independent step groups. The `identify_parallelizable_steps` method (lines 321-332) implements this:

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

# Build pipeline with mixed dependencies

builder = PipelineBuilder()
pipeline = (
    builder
    .add_step("ingest_a", "file_ingest", source="./folder_a")
    .add_step("ingest_b", "file_ingest", source="./folder_b")  # No dependency!

    .add_step("merge", "combine", dependencies=["ingest_a", "ingest_b"])
    .add_step("analyze", "sentiment", dependencies=["merge"])
    .set_parallelism(2)
    .build(name="mixed_deps")
)

# ExecutionEngine passes to ParallelismManager

engine = ExecutionEngine()
result = engine.execute_pipeline(pipeline)

# ingest_a and ingest_b execute concurrently (no mutual dependency)

# merge waits for both; analyze waits for merge

```

The dependency analysis respects all `dependencies` arrays declared in step definitions. Steps without inter-dependencies form parallelizable groups regardless of their original declaration order.

### Step 2: Concurrent Execution via Thread Pools

The `execute_parallel` method (lines 20-48) dispatches tasks using `ThreadPoolExecutor` by default, with optional process-based execution:

```python

# Simplified core execution flow from parallelism_manager.py

def execute_parallel(self, tasks, use_processes=False):
    """
    Execute tasks with configured parallelism level.
    
    Sorts by priority, then dispatches to thread or process pool.
    """
    sorted_tasks = sorted(tasks, key=lambda t: t.priority, reverse=True)
    
    if use_processes:
        return self._execute_with_processes(sorted_tasks)
    return self._execute_with_threads(sorted_tasks)  # default

def _execute_with_threads(self, tasks):
    from concurrent.futures import ThreadPoolExecutor, as_completed
    
    with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
        futures = {executor.submit(t.run): t for t in tasks}
        results = []
        for future in as_completed(futures):
            task = futures[future]
            try:
                result = future.result()
                results.append(TaskResult(task.task_id, True, result))
            except Exception as e:
                results.append(TaskResult(task.task_id, False, error=e))
        return results

```

The `_execute_with_threads` implementation (lines 71-77 in the source) provides:

- **Configurable worker count** — `max_workers` derived from pipeline metadata
- **Priority-aware scheduling** — higher-priority tasks submit first
- **Result aggregation** — collects success/failure status and return values
- **Exception isolation** — failures in one task don't crash the pool

### Step 3: Status Tracking and Completion

As steps complete, the `ExecutionEngine` updates each `PipelineStep`'s `status` attribute to `StepStatus.COMPLETED` or `StepStatus.FAILED`. This enables downstream steps to check dependency satisfaction before execution.

## Complete Working Example

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

def build_parallel_document_pipeline():
    """
    Demonstrates full parallel execution with the Semantica DSL.
    """
    builder = PipelineBuilder()
    
    # Three independent ingestion steps can run in parallel

    # Analysis depends on all three completing first

    pipeline = (
        builder
        .add_step("ingest_docs", "file_ingest", source="./documents/")
        .add_step("ingest_images", "file_ingest", source="./images/")
        .add_step("ingest_metadata", "file_ingest", source="./meta.json")
        .add_step(
            "parse_docs", 
            "document_parse", 
            dependencies=["ingest_docs"]
        )
        .add_step(
            "embed_all",
            "multi_modal_embed",
            dependencies=["parse_docs", "ingest_images", "ingest_metadata"]
        )
        .add_step(
            "index",
            "vector_index",
            dependencies=["embed_all"]
        )
        .set_parallelism(3)  # Run ingest_docs, ingest_images, ingest_metadata concurrently

        .build(name="document_processing")
    )
    
    engine = ExecutionEngine()
    result = engine.execute_pipeline(pipeline)
    
    print(f"Success: {result.success}")
    print(f"Steps executed: {result.metrics['steps_executed']}")
    print(f"Parallel groups: {result.metrics.get('parallel_groups', 'N/A')}")
    
    return result

if __name__ == "__main__":
    build_parallel_document_pipeline()

```

Execution flow for this pipeline:

1. **Group 1 (parallel)**: `ingest_docs`, `ingest_images`, `ingest_metadata` — 3 workers active
2. **Group 2 (sequential)**: `parse_docs` — waits for `ingest_docs` only
3. **Group 3 (sequential)**: `embed_all` — waits for all three ingestion steps plus parsing
4. **Group 4 (sequential)**: `index` — final step

## Advanced: Direct ParallelismManager Usage

For fine-grained control outside the DSL, instantiate `ParallelismManager` directly:

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

def parallel_url_fetch(urls):
    """
    Manual parallel task execution without Pipeline DSL.
    """
    def fetch(url):
        import requests
        return requests.get(url, timeout=30).status_code
    
    tasks = [
        Task(task_id=f"fetch_{i}", handler=fetch, args=(url,), priority=len(urls)-i)
        for i, url in enumerate(urls)
    ]
    
    manager = ParallelismManager(max_workers=4)
    results = manager.execute_parallel(tasks, use_processes=False)
    
    return {
        r.task_id: {
            "success": r.success,
            "result": r.result if r.success else str(r.error)
        }
        for r in results
    }

```

## Key Files and Their Responsibilities

| File | Primary Role |
|------|-------------|
| [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) | DSL implementation; `set_parallelism()` method; stores configuration |
| [`semantica/pipeline/parallelism_manager.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/parallelism_manager.py) | Dependency graph analysis; thread/process pool execution |
| [`semantica/pipeline/execution_engine.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/execution_engine.py) | Orchestrates pipeline runs; integrates `ParallelismManager` |
| [`semantica/pipeline/pipeline_templates.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_templates.py) | Pre-configured templates with parallelism presets |
| [`tests/test_pipeline_orchestration.py`](https://github.com/semantica-agi/semantica/blob/main/tests/test_pipeline_orchestration.py) | Unit tests for parallel step identification |
| [`tests/pipeline/test_pipeline_comprehensive.py`](https://github.com/semantica-agi/semantica/blob/main/tests/pipeline/test_pipeline_comprehensive.py) | End-to-end pipeline execution tests |

## Summary

- **Declarative parallelism** — Call `set_parallelism(level)` or include `"parallelism"` in config; the DSL stores this in `pipeline_config` until build time
- **Metadata propagation** — `PipelineBuilder.build()` attaches the parallelism level to `Pipeline.metadata` for runtime access
- **Automatic dependency analysis** — `ParallelismManager.identify_parallelizable_steps()` groups independent steps regardless of declaration order
- **Thread-pool execution** — `execute_parallel()` dispatches via `ThreadPoolExecutor` (default) or `ProcessPoolExecutor` when `use_processes=True`
- **Dependency respect** — Steps explicitly declare dependencies; the execution engine never violates these constraints, only exploits independence

## Frequently Asked Questions

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

Call `set_parallelism(n)` on your `PipelineBuilder` instance before invoking `build()`, or include `"parallelism": n` in your configuration dictionary. The value must be at least 1. The DSL validates and stores this in `pipeline_config`, then propagates it to the `Pipeline` metadata for the execution engine to consume.

### Does parallelism respect step dependencies?

Yes. `ParallelismManager` explicitly analyzes the dependency graph via `identify_parallelizable_steps()` (lines 321-332 in [`parallelism_manager.py`](https://github.com/semantica-agi/semantica/blob/main/parallelism_manager.py)). Only steps without mutual dependencies execute concurrently. If step B depends on step A, B always waits for A's completion regardless of the parallelism level.

### Can I use process pools instead of threads?

Yes. Pass `use_processes=True` when calling `ParallelismManager.execute_parallel()` directly. The DSL does not expose this flag in `set_parallelism()`, but the manager supports both `_execute_with_threads()` (default) and `_execute_with_processes()` internally depending on this parameter.

### What happens if a parallel step fails?

The `ThreadPoolExecutor` isolates exceptions per task. `ParallelismManager` catches failures, records `StepStatus.FAILED` in the result, and continues processing other concurrent tasks. The `ExecutionEngine` then determines whether to fail the entire pipeline or continue based on configured error handling policies.