semantica/docs/reference/pipeline.md

title, description, icon

title	description	icon
Pipeline Module	Pipeline DSL with parallel workers, retry policies, failure handling, and progress tracking.	gear

Robust orchestration engine for building and executing complex data processing workflows.

---

Overview

The Pipeline Module lets you chain Semantica components into reproducible, fault-tolerant workflows with parallel execution and configurable error handling.

---

Basic Pipeline

from semantica.pipeline import Pipeline
from semantica.ingest import FileIngestor
from semantica.parse import DocumentParser
from semantica.semantic_extract import NERExtractor
from semantica.kg import GraphBuilder

pipeline = Pipeline()
pipeline.add_step("ingest",   FileIngestor())
pipeline.add_step("parse",    DocumentParser())
pipeline.add_step("extract",  NERExtractor(method="llm", llm_provider=llm))
pipeline.add_step("build_kg", GraphBuilder(merge_entities=True))

result = pipeline.run("data/")
kg = result.output

---

Parallel Processing

pipeline = Pipeline(workers=4)   # run steps in parallel across documents

pipeline.add_step("ingest",  FileIngestor())
pipeline.add_step("parse",   DocumentParser())
pipeline.add_step("extract", NERExtractor(), parallel=True, batch_size=10)
pipeline.add_step("build",   GraphBuilder())

result = pipeline.run("data/")

---

Retry & Error Handling

from semantica.pipeline import Pipeline, RetryPolicy, FailureHandler

retry = RetryPolicy(
    max_retries=3,
    backoff="exponential",   # "fixed" | "linear" | "exponential"
    initial_delay=1.0
)

handler = FailureHandler(
    strategy="skip",         # "skip" | "stop" | "retry"
    log_failures=True
)

pipeline = Pipeline(retry_policy=retry, failure_handler=handler)

---

Progress Tracking

# Print progress to console
result = pipeline.run("data/", show_progress=True)

# WebSocket progress (via Knowledge Explorer)
result = pipeline.run("data/", websocket_port=8080)

print(f"Processed: {result.processed_count}")
print(f"Failed: {result.failed_count}")
print(f"Duration: {result.duration_seconds:.1f}s")

---

Pipeline DSL

from semantica.pipeline import PipelineBuilder

pipeline = (
    PipelineBuilder()
    .ingest(FileIngestor())
    .parse(DocumentParser())
    .normalize()
    .extract(NERExtractor(method="llm", llm_provider=llm))
    .extract_relations(RelationExtractor(method="llm", llm_provider=llm))
    .build_kg(merge_entities=True)
    .deduplicate(strategy="semantic_v2")
    .export(format="turtle", path="output.ttl")
    .build()
)

result = pipeline.run("data/")

---

Saving & Loading Pipelines

# Save pipeline definition
pipeline.save("pipeline_config.yaml")

# Load and run
pipeline = Pipeline.load("pipeline_config.yaml")
result = pipeline.run("data/")

---

See Also

First step in most pipelines. Core extraction step. Graph construction step. Final output step.