How the PipelineBuilder DSL Handles Parallel Execution in Semantica

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 (lines 175-186). This stores the desired worker count in the builder’s internal pipeline_config dictionary:

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 (lines 221-224), the implementation embeds the configuration as follows:

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

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:

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:

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.
• 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, 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 (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). 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). Steps that do not depend on each other’s outputs are grouped together and executed concurrently, while maintaining sequential execution for dependent steps.

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