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

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

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). This adds the source step name to the target's dependencies list.

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

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

Convert the declarative description into an executable Pipeline object:

# Build from fluent API calls

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

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

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), 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)

Controlling Parallelism

Independent steps—those without mutual dependencies—can execute concurrently. Control worker allocation with set_parallelism() (lines 87-101):

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

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

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

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	Core DSL, step handling, validation orchestration, and serialization entry points
semantica/pipeline/pipeline_validator.py	Topology validation: missing steps, circular dependencies, invalid references
semantica/pipeline/pipeline_serializer.py	JSON and custom format serialization via nested PipelineSerializer class
semantica/utils/progress_tracker.py	Progress reporting during build and validation phases
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 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().

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