How to Compose Ingestion Pipelines with Semantica's PipelineBuilder

Use the PipelineBuilder class from 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 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, 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 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 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.

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 catches structural errors early.

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

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.

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:

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:

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.

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