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

semantica.pipeline lets you chain Semantica components into reproducible, fault-tolerant workflows with parallel execution and configurable error handling. Pipelines are serializable — save them to YAML and reload in any environment.

What You Get

• Pipeline — chain steps with parallel workers, retry policies, and failure handlers
• PipelineBuilder — fluent DSL for building pipelines with a readable chain syntax
• RetryPolicy — fixed, linear, and exponential backoff with configurable max retries
• FailureHandler — skip, stop, or retry failed documents without halting the pipeline
• Progress tracking — console (tqdm), WebSocket streaming, or file logging

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

Process documents concurrently across multiple workers:

pipeline = Pipeline(workers=4)

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 and Error Handling

Configure retry behavior and failure strategy independently:

from semantica.pipeline import Pipeline, RetryPolicy, FailureHandler

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

handler = FailureHandler(
    strategy="skip",         # "skip" | "stop" | "retry"
    log_failures=True        # write failed documents to error log
)

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

Progress Tracking

# Console progress bar (tqdm)
result = pipeline.run("data/", show_progress=True)

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

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

Pipeline DSL

The PipelineBuilder provides a fluent chain syntax that reads as a data flow:

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/")

Save and Load Pipelines

Serialize a pipeline to YAML for reproducible runs across environments:

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

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

Pipeline Result

@dataclass
class PipelineResult:
    output:           Any      # final step output (e.g., a KnowledgeGraph)
    processed_count:  int      # documents successfully processed
    failed_count:     int      # documents that failed after retries
    duration_seconds: float    # total wall-clock time
    step_metrics:     Dict     # per-step timing and counts
    errors:           List     # list of FailedDocument records

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