# How to Create Declarative Pipelines with Semantica's PipelineBuilder: A Complete Guide

> Learn to create declarative pipelines with Semantica's PipelineBuilder. This guide shows how to chain steps, wire dependencies, and build validated pipelines easily.

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

---

**Semantica's `PipelineBuilder` provides a fluent, declarative API for describing data-processing workflows by chaining step declarations, wiring dependencies, and materializing validated `Pipeline` objects.**

The `PipelineBuilder` class in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) separates **what** your pipeline does from **how** it runs. This declarative approach enables reproducible workflows, version-controlled configurations, and easier testing of complex data processing pipelines.

## Core Concepts: Steps, Dependencies, and Materialization

A declarative pipeline consists of three fundamental elements:

- **Steps** – Named units of work with a type that maps to a handler function
- **Dependencies** – Directed edges that enforce execution order
- **Configuration** – Step-specific parameters and global settings like parallelism

The builder implements this through a three-stage workflow that keeps description isolated from execution.

## Step 1: Declare Pipeline Steps with `add_step()`

Register each step using `add_step(name, type, **config)` (lines [20-31](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py#L20-L31)). This creates a `PipelineStep` dataclass instance that stores:

- `name` – Unique identifier for the step
- `type` – String key mapping to a runtime handler
- `config` – Arbitrary parameters passed to the handler
- `dependencies` – List of prerequisite step names
- `parallel_safe` – Boolean flag for concurrency

```python
from semantica.pipeline.pipeline_builder import PipelineBuilder

builder = PipelineBuilder()

# Declare steps with optional configuration

builder.add_step("ingest", "file_ingestor", path="/data/input.csv")
builder.add_step("clean", "data_cleaner", clean_missing=True)
builder.add_step("embed", "text_embedder", model="all-MiniLM-L6-v2")
builder.add_step("store", "vector_store", store="qdrant")

```

Each step is independent at declaration time. Dependencies are wired separately, keeping the API flexible for dynamic pipeline construction.

## Step 2: Wire Dependencies with `connect_steps()`

Establish execution order using `connect_steps(source, target)` (lines [64-84](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py#L64-L84)). This adds the source step name to the target's `dependencies` list.

```python

# Linear dependency chain: ingest → clean → embed → store

builder.connect_steps("ingest", "clean")
builder.connect_steps("clean", "embed")
builder.connect_steps("embed", "store")

```

The builder validates these connections during materialization. Circular dependencies and missing step references raise `ValidationError` from [`semantica/utils/exceptions.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/utils/exceptions.py).

## Step 3: Materialize with `build()` or `build_pipeline()`

Convert the declarative description into an executable `Pipeline` object:

```python

# Build from fluent API calls

pipeline = builder.build(name="text_processing_pipeline")

```

For external configuration, use `build_pipeline()` with a dictionary:

```python
pipeline_cfg = {
    "name": "yaml_defined_pipeline",
    "steps": [
        {"name": "ingest", "type": "file_ingestor", "config": {"path": "/data/input.csv"}},
        {"name": "clean", "type": "data_cleaner", "config": {"clean_missing": True},
         "dependencies": ["ingest"]},
        {"name": "embed", "type": "text_embedder", "config": {"model": "all-MiniLM-L6-v2"},
         "dependencies": ["clean"]},
        {"name": "store", "type": "vector_store", "config": {"store": "qdrant"},
         "dependencies": ["embed"]},
    ],
    "parallelism": 2,
}

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

```

During `build()` (lines [125-134](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py#L125-L134)), the builder:

1. Invokes `PipelineValidator` to check topology
2. Assembles metadata including step count and parallelism level
3. Returns a fully-typed `Pipeline` dataclass (lines [73-84](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py#L73-L84))

## Controlling Parallelism

Independent steps—those without mutual dependencies—can execute concurrently. Control worker allocation with `set_parallelism()` (lines [87-101](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py#L87-L101)):

```python
builder.set_parallelism(level=4)

```

The runtime engine uses this value to schedule parallel-safe steps across the available worker pool.

## Serializing and Persisting Pipelines

Save pipeline definitions for version control or distribution using `serialize()` (lines [155-176](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py#L155-L176)):

```python

# Export to JSON

json_repr = builder.serialize(format="json")

# Restore later

from semantica.pipeline.pipeline_builder import PipelineSerializer

serializer = PipelineSerializer()
restored_pipeline = serializer.deserialize_pipeline(json_repr)

```

The `PipelineSerializer` class (starting at line [165](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py#L165)) handles schema validation during deserialization, ensuring reconstructed pipelines match the original specification.

## Early Validation with `validate_pipeline()`

Catch configuration errors before materialization using `validate_pipeline()` (lines [188-208](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py#L188-L208)):

```python
builder.validate_pipeline()  # Raises ValidationError on problems

```

This runs the same checks as `build()` without constructing the final `Pipeline` object, useful for CI/CD pipelines or interactive development.

## Complete PipelineBuilder Example

```python
from semantica.pipeline.pipeline_builder import PipelineBuilder, PipelineSerializer

# Initialize builder

builder = PipelineBuilder()

# 1. Declare all steps

builder.add_step("ingest", "file_ingestor", path="/data/input.csv", parallel_safe=True)
builder.add_step("clean", "data_cleaner", clean_missing=True)
builder.add_step("embed", "text_embedder", model="all-MiniLM-L6-v2")
builder.add_step("store", "vector_store", store="qdrant")

# 2. Wire execution order

builder.connect_steps("ingest", "clean")
builder.connect_steps("clean", "embed")
builder.connect_steps("embed", "store")

# 3. Configure parallelism (ingest and any other independent steps can run together)

builder.set_parallelism(level=4)

# 4. Build and validate

pipeline = builder.build(name="semantic_search_pipeline")

# 5. Inspect results

print(f"Steps: {pipeline.metadata.step_count}")
print(f"Parallelism: {pipeline.metadata.parallelism}")

# 6. Serialize for deployment

config_json = builder.serialize(format="json")

```

## Key Files in the Pipeline System

| File | Purpose |
|------|---------|
| [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) | Core DSL, step handling, validation orchestration, and serialization entry points |
| [`semantica/pipeline/pipeline_validator.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_validator.py) | Topology validation: missing steps, circular dependencies, invalid references |
| [`semantica/pipeline/pipeline_serializer.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_serializer.py) | JSON and custom format serialization via nested `PipelineSerializer` class |
| [`semantica/utils/progress_tracker.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/utils/progress_tracker.py) | Progress reporting during build and validation phases |
| [`semantica/utils/exceptions.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/utils/exceptions.py) | `ProcessingError` and `ValidationError` exception definitions |

## Summary

- **Declarative pipelines** in Semantica separate workflow description from execution through the `PipelineBuilder` class
- **`add_step()`** registers work units; **`connect_steps()`** establishes execution order
- **`build()`** materializes validated `Pipeline` objects; **`build_pipeline()`** accepts dictionary configurations
- **`set_parallelism()`** controls concurrent execution of independent steps
- **`serialize()`** and **`PipelineSerializer.deserialize_pipeline()`** enable persistence and deployment
- **`validate_pipeline()`** provides early error detection without full materialization

## Frequently Asked Questions

### What is the difference between `build()` and `build_pipeline()`?

`build()` materializes a pipeline from fluent API calls made on the builder instance. `build_pipeline()` accepts a dictionary configuration and internally populates the builder state before materialization. Both produce identical `Pipeline` objects and run the same validation logic via `PipelineValidator`.

### How does Semantica handle circular dependencies in pipelines?

The `PipelineValidator` (invoked during `build()` or `validate_pipeline()`) performs a cycle detection pass on the dependency graph. If a circular reference is found, it raises `ValidationError` from [`semantica/utils/exceptions.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/utils/exceptions.py) before any `Pipeline` object is constructed.

### Can I modify a pipeline after calling `build()`?

No. `Pipeline` objects returned by `build()` are immutable dataclasses. To change a pipeline, create a new `PipelineBuilder` instance, re-declare steps with modifications, and call `build()` again. Alternatively, deserialize a previously serialized configuration, modify the dictionary, and pass it to `build_pipeline()`.