How to Use PipelineBuilder for Declarative ETL Workflows in Semantica

The PipelineBuilder class in semantica.pipeline provides a fluent DSL for constructing validated, serializable ETL pipelines by chaining steps with explicit dependencies, configuration, and optional version-control metadata.

The semantica-agi/semantica repository offers a powerful toolkit for defining complex data workflows declaratively. Using the PipelineBuilder class located in semantica/pipeline/pipeline_builder.py, you can assemble ETL processes as immutable directed acyclic graphs (DAGs) without writing imperative execution logic.

Core Architecture of the Pipeline Builder

The builder operates on a declarative paradigm: you describe what the workflow looks like, not how the engine executes it. The primary components include:

• PipelineBuilder – Provides the fluent API to add steps, wire dependencies, configure parallelism, and produce a Pipeline dataclass. Implemented in semantica/pipeline/pipeline_builder.py.
• Pipeline & PipelineStep – Immutable dataclasses storing step names, types, configurations, and dependency graphs.
• PipelineValidator – Validates structural integrity, detects circular dependencies, and runs performance heuristics. Located in semantica/pipeline/pipeline_validator.py.
• PipelineSerializer – Handles serialization to JSON/dict formats for reproducibility and remote execution.

Once built, a Pipeline instance can be validated for errors, serialized for storage, or passed to the ExecutionEngine for actual processing.

Defining Steps and Dependencies

The add_step() method creates individual pipeline stages with unique identifiers and explicit predecessors.

from semantica.pipeline import PipelineBuilder

builder = PipelineBuilder()
pipeline = (
    builder
    .add_step("ingest", "ingest", source="./documents")
    .add_step(
        "parse",
        "parse",
        formats=["pdf", "docx"],
        dependencies=["ingest"]
    )
    .add_step(
        "embed",
        "embed",
        model="text-embedding-3-large",
        batch_size=32,
        dependencies=["parse"]
    )
    .build(name="DocumentETL")
)

Key parameters include:

• name – Unique step identifier enforced across the pipeline.
• type – Logical operation category (e.g., ingest, parse, embed).
• dependencies – List of predecessor step names defining execution order.
• handler – Optional callable performing the actual work.
• delta_mode, base_version_id, target_version_id – Optional version-control flags for tracking data lineage.

For dynamic wiring after step creation, use connect_steps(from_step, to_step) to append dependencies programmatically.

Validation and Serialization

Before execution, validate the pipeline structure to catch configuration errors early:

validation = builder.validate_pipeline()
if not validation["valid"]:
    raise RuntimeError(f"Pipeline validation failed: {validation['errors']}")

The PipelineValidator checks for duplicate step names, missing dependencies, circular references, and performance anti-patterns. According to the source code in semantica/pipeline/pipeline_validator.py, it returns a dictionary with valid, errors, and warnings keys.

Serialize pipelines for reproducibility or remote deployment:

from semantica.pipeline import PipelineSerializer

serializer = PipelineSerializer()
json_repr = serializer.serialize_pipeline(pipeline, format="json")
restored = serializer.deserialize_pipeline(json_repr)

Practical Implementation Examples

Basic Fluent API Usage

Construct a complete ETL workflow using method chaining. This example implements a document processing pipeline with normalization and entity extraction:

from semantica.pipeline import PipelineBuilder, ExecutionEngine

builder = PipelineBuilder()
pipeline = (
    builder
    .add_step("ingest", "ingest", source="./documents")
    .add_step(
        "parse",
        "parse",
        formats=["pdf", "docx"],
        dependencies=["ingest"]
    )
    .add_step(
        "normalize",
        "normalize",
        case="lower",
        dependencies=["parse"]
    )
    .add_step(
        "extract",
        "extract",
        entities=True,
        dependencies=["normalize"]
    )
    .add_step(
        "embed",
        "embed",
        model="text-embedding-3-large",
        batch_size=32,
        dependencies=["extract"]
    )
    .build(name="DocumentETL")
)

# Execute

engine = ExecutionEngine()
result = engine.execute_pipeline(pipeline)
print(f"Success: {result.success}")

Building from Declarative Configuration

Load pipeline definitions from YAML or JSON files using build_pipeline():

pipeline_cfg = {
    "name": "YAMLBasedPipeline",
    "steps": [
        {
            "name": "ingest",
            "type": "ingest",
            "config": {"source": "./data"}
        },
        {
            "name": "parse",
            "type": "parse",
            "config": {"formats": ["txt"]},
            "dependencies": ["ingest"]
        },
        {
            "name": "embed",
            "type": "embed",
            "config": {"model": "text-embedding-3-large"},
            "dependencies": ["parse"]
        },
    ],
    "parallelism": 2,
}

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

The build_pipeline() method (lines 41-90 in pipeline_builder.py) automatically materializes steps from the dictionary structure.

Using Pipeline Templates

Leverage pre-built templates for common workflows:

from semantica.pipeline import PipelineTemplateManager

tmpl_mgr = PipelineTemplateManager()
builder = tmpl_mgr.create_pipeline_from_template(
    "document_processing",
    ingest={"source": "./docs"},
    parse={"formats": ["pdf", "docx"]},
    embed={"model": "text-embedding-3-large"},
)
pipeline = builder.build()

Template definitions reside in semantica/pipeline/pipeline_templates.py, with usage documentation available in semantica/pipeline/pipeline_usage.md.

Configuring Parallelism

Set global execution concurrency hints using set_parallelism():

builder = PipelineBuilder()
builder.set_parallelism(level=4)

This value instructs the execution engine how many worker threads or processes to utilize when running independent steps.

Summary

• PipelineBuilder in semantica/pipeline/pipeline_builder.py enables declarative ETL construction through a fluent interface.
• Steps are defined via add_step() with unique names, types, and optional dependencies lists.
• Validation through validate_pipeline() catches structural errors, circular dependencies, and missing references before runtime.
• Serialization via PipelineSerializer supports JSON output for reproducible pipeline storage and remote execution.
• Templates via PipelineTemplateManager provide pre-configured workflows for common patterns like document processing.
• Configuration-driven workflows can be instantiated from dictionaries using build_pipeline().

Frequently Asked Questions

What is the difference between PipelineBuilder and ExecutionEngine?

PipelineBuilder constructs and validates the static pipeline definition (the DAG), while ExecutionEngine consumes that definition and orchestrates the actual runtime execution, handling resource scheduling and parallelism.

How do I detect circular dependencies before runtime?

Call builder.validate_pipeline() before execution. The PipelineValidator class in semantica/pipeline/pipeline_validator.py performs topological sorting and detects cycles, returning specific error messages identifying the problematic dependency chain.

Can I serialize a pipeline to JSON and restore it later?

Yes. Use PipelineSerializer.serialize_pipeline(pipeline, format="json") to generate a JSON string, then call deserialize_pipeline() to reconstruct the Pipeline dataclass. This enables version-controlled workflow definitions and distributed execution.

How does version control (delta mode) work in pipeline steps?

When adding steps, you can specify delta_mode=True along with base_version_id and target_version_id parameters. These flags are stored in the step configuration and consumed by downstream components in semantica/pipeline/pipeline_provenance.py to track data lineage and incremental processing states.

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