# Internal Architecture of Semantica's PipelineBuilder DSL: Core Components and Data Flow

> Explore Semantica's PipelineBuilder DSL internal architecture. Understand core components like Pipeline, PipelineStep, and PipelineBuilder for declarative DAG workflow creation and execution.

- Repository: [Semantica /semantica](https://github.com/semantica-agi/semantica)
- Tags: internals
- Published: 2026-09-11

---

**Semantica's PipelineBuilder DSL provides a fluent interface for constructing, validating, and executing directed acyclic graph (DAG) workflows through three core data structures—`Pipeline`, `PipelineStep`, and `PipelineBuilder`—enabling declarative pipeline definition with static validation and extensible step handlers.**

The `semantica-agi/semantica` repository implements a powerful Domain-Specific Language (DSL) for orchestrating knowledge-graph and vector-store workflows. This internal architecture analysis examines how the **PipelineBuilder DSL** transforms declarative pipeline definitions into executable DAGs through its layered validation, serialization, and execution systems.

## Core Data Model of the PipelineBuilder DSL

### The Pipeline Dataclass

In [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py), the `Pipeline` dataclass serves as the immutable container for workflow definitions. It stores a list of `PipelineStep` objects, global configuration parameters, and runtime metadata required for execution tracking and status monitoring across distributed environments.

### PipelineStep and Execution Status

Each `PipelineStep` instance represents a discrete operation within the DAG, encapsulating the step name, type identifier, configuration dictionary, downstream connections, and current execution state via the `StepStatus` enumeration. This structure enables the `ExecutionEngine` to traverse dependency graphs and manage step lifecycle transitions from pending through completion or failure states.

### PipelineBuilder as DSL Entry Point

The `PipelineBuilder` class provides the fluent API surface for constructing pipelines programmatically. According to the source code in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py), it exposes methods including `add_step()`, `connect_steps()`, `set_parallelism()`, and `register_step_handler()` that ultimately produce a validated `Pipeline` object through the `build()` method. This builder pattern abstracts the complexity of DAG construction while maintaining type safety and structural constraints.

## Validation and Serialization Infrastructure

### Structural Validation Layer

Before a pipeline becomes executable, [`semantica/pipeline/pipeline_validator.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_validator.py) performs static analysis to enforce structural integrity. The validation logic checks for orphaned steps without connections, validates parallelism constraints against hardware limits, and ensures all referenced step types have registered handlers, catching configuration errors prior to runtime deployment.

### Cross-Environment Persistence

The `PipelineSerializer` class in [`semantica/pipeline/pipeline_serializer.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_serializer.py) handles bidirectional conversion between `Pipeline` objects and JSON/YAML representations. This enables version-controlled pipeline definitions that can be stored in object storage, transmitted across environments, and reconstructed identically via deserialization methods for reproducible execution.

## Execution Engine and Handler Registration

### DAG Traversal in ExecutionEngine

Located in [`semantica/core/orchestrator.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/core/orchestrator.py), the `ExecutionEngine` consumes built `Pipeline` objects and orchestrates their execution. It walks the DAG of `PipelineStep` instances in topological order, invoking appropriate handlers while respecting parallelism constraints defined during the building phase through `set_parallelism()`.

### Extensible Step Handlers

The DSL supports custom logic injection through the `PipelineBuilder.register_step_handler(step_type, handler)` method. Developers override default behaviors to implement domain-specific operations such as vector-store ingestion, knowledge-graph analysis, or external API integration without modifying core framework code in the `semantica` package.

## Template System and Common Patterns

Beyond manual construction, [`semantica/pipeline/pipeline_templates.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_templates.py) provides pre-configured templates for recurring workflow patterns including "construct-template", "vector-store ingestion", and "knowledge-graph enrichment". These templates accelerate development by bundling validated step sequences with optimized default configurations that comply with the DSL's structural requirements.

## Practical Implementation Examples

### Basic Two-Step Pipeline Construction

```python
from semantica.pipeline import PipelineBuilder

builder = PipelineBuilder()
builder.add_step("load_data", "vector_store_ingest", source="s3://my-bucket/data")
builder.add_step("embed", "embedding", model="openai/text-embedding-ada-002")
builder.connect_steps("load_data", "embed")   # load_data → embed

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

```

### Parallel Execution with Custom Handlers

```python
from semantica.pipeline import PipelineBuilder

def my_custom_handler(step_config, context):
    # … user‑defined logic …

    return {"status": "ok"}

builder = PipelineBuilder()
builder.register_step_handler("my_custom", my_custom_handler)

builder.add_step("step_a", "my_custom", param=1)
builder.add_step("step_b", "my_custom", param=2)
builder.set_parallelism(level=2)                # run step_a and step_b concurrently

pipeline = builder.build("parallel_demo")

```

### Serialization and Deserialization Workflows

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

builder = PipelineBuilder()
builder.add_step("extract", "semantic_extractor", model="gpt-4")
pipeline = builder.build("exportable")

# Serialize

json_repr = PipelineSerializer.serialize(pipeline, format="json")
print(json_repr)

# Deserialize back to a Pipeline object

restored = PipelineSerializer.deserialize(json_repr, format="json")

```

## Summary

- The PipelineBuilder DSL centers on three core structures: `Pipeline`, `PipelineStep`, and `PipelineBuilder` defined in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py).
- Static validation occurs through [`pipeline_validator.py`](https://github.com/semantica-agi/semantica/blob/main/pipeline_validator.py) before execution to prevent runtime structural failures and orphaned steps.
- The `ExecutionEngine` in [`semantica/core/orchestrator.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/core/orchestrator.py) traverses the DAG and invokes registered step handlers in topological order.
- Custom handlers enable extensibility without core code modification via `register_step_handler()`.
- JSON/YAML serialization in [`pipeline_serializer.py`](https://github.com/semantica-agi/semantica/blob/main/pipeline_serializer.py) supports reproducible, version-controlled pipelines across environments.

## Frequently Asked Questions

### What are the three core data structures in Semantica's PipelineBuilder DSL?

The architecture relies on the `Pipeline` dataclass (container for workflow definitions), `PipelineStep` (individual operations with connection metadata and `StepStatus`), and `PipelineBuilder` (fluent API entry point), all implemented in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py). These structures work together to represent the DAG topology and execution state.

### How does the ExecutionEngine process a built pipeline?

The `ExecutionEngine` located in [`semantica/core/orchestrator.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/core/orchestrator.py) receives a validated `Pipeline` object, performs topological sorting of the `PipelineStep` nodes, and sequentially invokes registered handlers while respecting parallelism settings configured via `PipelineBuilder.set_parallelism()`. It manages step lifecycle transitions and tracks execution status throughout the workflow run.

### Can custom step handlers be registered in the PipelineBuilder DSL?

Yes. Developers call `PipelineBuilder.register_step_handler(step_type, handler)` to map custom Python functions to specific step type identifiers, enabling integration of vector-store operations, knowledge-graph algorithms, or external API calls without modifying the framework's core execution logic in the `semantica` package.

### Where is pipeline validation logic implemented in the codebase?

Structural validation logic resides in [`semantica/pipeline/pipeline_validator.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_validator.py), which checks for orphaned steps, validates that parallelism levels are correctly configured, and verifies that all step types have associated handlers before the `PipelineBuilder.build()` method returns a concrete `Pipeline` instance.