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

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): 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): 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): 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): 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 around line 175:

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:

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:

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:

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:

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

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