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

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. This method stores the desired worker count in the builder's internal pipeline_config dictionary.

# 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:

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 (lines 221-224):

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). 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:

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:

# 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

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:

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	DSL implementation; set_parallelism() method; stores configuration
semantica/pipeline/parallelism_manager.py	Dependency graph analysis; thread/process pool execution
semantica/pipeline/execution_engine.py	Orchestrates pipeline runs; integrates ParallelismManager
semantica/pipeline/pipeline_templates.py	Pre-configured templates with parallelism presets
tests/test_pipeline_orchestration.py	Unit tests for parallel step identification
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). 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}