# How to Compose Ingestion Pipelines with Semantica's PipelineBuilder

> Learn to compose ingestion pipelines using Semantica's PipelineBuilder. Declaratively build DAGs, validate, and execute with dependency resolution and parallelism.

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

---

**Use the `PipelineBuilder` class from [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) to declaratively construct directed acyclic graphs of processing steps, then validate and execute them with automatic dependency resolution, parallelism, and optional provenance tracking.**

Ingestion pipelines in Semantica transform raw data into vector-ready embeddings through modular, composable steps. The `PipelineBuilder` API abstracts away execution complexity—dependency wiring, retries, timeouts, and parallel scheduling—while keeping each step testable in isolation. This guide walks through building, validating, and running pipelines using the actual source implementation from the `semantica-agi/semantica` repository.

## Core Components of Semantica's Pipeline System

Understanding the architecture helps you compose pipelines effectively. The system separates definition, validation, and execution into distinct layers.

### PipelineBuilder: The Entry Point

The `PipelineBuilder` class in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) provides a fluent interface for registering steps. Each step wraps a Python callable with metadata controlling its runtime behavior.

Key methods:
- `add_step(name, callable, **options)` — registers a processing function
- `build(name)` — materializes a runnable `Pipeline` object

Options include:
- `requires` — list of step names whose outputs feed this step
- `retry` — maximum retry attempts (default: 0)
- `timeout` — per-step timeout in seconds

### PipelineStep: Step-Level Configuration

Defined in [`semantica/pipeline/pipeline_step.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_step.py), the `PipelineStep` class encapsulates:
- The user-provided callable
- Dependency declarations (`requires`)
- Retry policies and timeout limits
- Input/output type hints (when provided)

Steps form nodes in the execution DAG. The builder validates that the graph remains acyclic and that all required inputs are satisfied before construction.

### ExecutionEngine: DAG Execution

The `ExecutionEngine` in [`semantica/pipeline/execution_engine.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/execution_engine.py) handles:
- **Dependency resolution**: topological sort of the step graph
- **Parallel scheduling**: independent branches execute concurrently
- **Error propagation**: failures halt dependent steps, optional retry logic applied
- **Result injection**: outputs of completed steps automatically become inputs to downstream steps

### PipelineWithProvenance: Observability Layer

For audit and reproducibility, [`semantica/pipeline/pipeline_provenance.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_provenance.py) provides a wrapper that records:
- Step start/end timestamps
- Input and output values (configurable)
- Exception traces for failed steps

Enable by wrapping your built pipeline before execution.

## Building a Basic Ingestion Pipeline

This example demonstrates a four-step pipeline: load raw decisions, parse them into structured objects, generate embeddings, and store to a vector database.

```python
from semantica.pipeline import PipelineBuilder, PipelineValidator
from semantica.vector_store import VectorStore
from semantica.context import DecisionContext

# Initialize builder

builder = PipelineBuilder()

# Step 1: Load raw decision data from external source

def load_decisions(_):
    # Implementation: API call, file read, or database query

    return [{"decision_id": 1, "text": "Approve loan application"}, ...]

builder.add_step(
    name="load",
    callable=load_decisions,
    retry=3,
    timeout=60,
)

# Step 2: Parse into DecisionContext objects

def parse_decisions(raw_decisions):
    return [DecisionContext(**d) for d in raw_decisions]

builder.add_step(
    name="parse",
    callable=parse_decisions,
    requires=["load"],  # consumes output of 'load' step

)

# Step 3: Generate embeddings using Semantica's embedding pipeline

def embed_decisions(contexts):
    from semantica.vector_store import DecisionEmbeddingPipeline
    pipeline = DecisionEmbeddingPipeline()
    return [pipeline.process_decision(c) for c in contexts]

builder.add_step(
    name="embed",
    callable=embed_decisions,
    requires=["parse"],
    retry=2,
)

# Step 4: Persist to vector store

def store_embeddings(embeddings):
    store = VectorStore.get_default()
    store.upsert(embeddings)

builder.add_step(
    name="store",
    callable=store_embeddings,
    requires=["embed"],
)

```

## Validating and Executing Your Pipeline

Always validate before building. The `PipelineValidator` in [`semantica/pipeline/pipeline_validator.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_validator.py) catches structural errors early.

```python

# Validate DAG structure

validator = PipelineValidator()
validator.validate(builder)  # raises PipelineValidationError on issues

# Build runnable pipeline

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

# Execute with optional initial context

result = pipeline.run(initial_context={})
print(f"Pipeline success: {result.success}")

```

Validation checks:
- No cycles in the dependency graph
- All `requires` references resolve to defined steps
- No duplicate step names
- Timeout and retry values are positive integers

## Enabling Provenance Tracking

Capture execution metadata for debugging and compliance:

```python
from semantica.pipeline import PipelineWithProvenance

# Wrap built pipeline

tracked = PipelineWithProvenance(pipeline, provenance=True)

result = tracked.run({})

# Access detailed execution log

for entry in result.provenance.step_log:
    print(f"{entry.step_name}: {entry.duration_ms}ms, status={entry.status}")

```

The provenance log includes timing, input/output snapshots (when configured), and retry attempts.

## Template-Driven Pipeline Construction

For reusable, versioned pipelines, use `PipelineTemplateManager` from [`semantica/pipeline/pipeline_template_manager.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_template_manager.py).

```python
from semantica.pipeline import PipelineTemplateManager

# Load predefined template

mgr = PipelineTemplateManager()
template = mgr.load_template("default_ingest.yaml")

# Scaffold builder from template

builder = template.scaffold()

# Extend or override specific steps

builder.add_step(
    name="custom_enrich",
    callable=add_metadata,
    requires=["embed"],
)

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

```

Template files define step skeletons—names, dependencies, and default options—while leaving callables to be bound at scaffolding time or overridden post-load.

## Design Patterns for Complex Pipelines

### Branching and Merging

The DAG structure supports independent branches that merge downstream:

```python
builder.add_step(name="fetch_text", callable=get_text, requires=["load"])
builder.add_step(name="fetch_images", callable=get_images, requires=["load"])

builder.add_step(
    name="multimodal_embed",
    callable=combine_embeddings,
    requires=["fetch_text", "fetch_images"],  # waits for both branches

)

```

The `ExecutionEngine` automatically parallelizes `fetch_text` and `fetch_images`.

### Conditional Steps

Implement conditional logic within step callables. The builder itself does not support conditional edges—use early returns or no-op outputs instead:

```python
def maybe_filter(data, context):
    if context.get("skip_filter"):
        return data  # pass-through

    return [d for d in data if d.score > 0.5]

builder.add_step(name="filter", callable=maybe_filter, requires=["parse"])

```

## Summary

- **Instantiate** `PipelineBuilder` to begin composing ingestion pipelines with Semantica
- **Add steps** with `add_step()`, declaring dependencies via `requires` and reliability settings via `retry` and `timeout`
- **Validate** with `PipelineValidator` to catch cycles and missing dependencies before runtime
- **Build** with `builder.build(name)` to produce a runnable `Pipeline` conforming to the `run(context)` protocol
- **Execute** via `pipeline.run()`—the `ExecutionEngine` handles parallelism, injection, and error handling automatically
- **Track provenance** by wrapping with `PipelineWithProvenance` for audit logs and reproducibility
- **Use templates** via `PipelineTemplateManager` for reusable, versioned pipeline definitions

## Frequently Asked Questions

### How do I specify that a step depends on multiple previous steps?

Pass a list to the `requires` parameter: `requires=["step_a", "step_b"]`. The `ExecutionEngine` will wait for both steps to complete before scheduling the dependent step, and their outputs will be available as a combined input.

### Can I modify a pipeline after building it?

No—`builder.build()` produces an immutable `Pipeline` object. To make changes, return to the `PipelineBuilder` instance (if preserved), or scaffold a new builder from a template and reapply your modifications.

### What happens when a step exceeds its timeout or retry limit?

The `ExecutionEngine` raises a `PipelineExecutionError` containing the step name and failure details. By default, this halts the entire pipeline. Implement custom error handling within your callable, or use `try/except` blocks to return partial results and allow downstream steps to continue.

### Does Semantica support dynamic step addition during execution?

No—the DAG must be fully defined at build time. For dynamic workflows, construct multiple pipeline variants and select between them at runtime, or implement conditional logic within step callables rather than conditional graph structure.