Pipeline API

scinr.newton.pipeline.run_pipeline async

run_pipeline(
    input_raw: str | None = None,
    converter_output_dir: str | None = None,
    extraction_input_dir: str | None = None,
    extraction_output_dir: str | None = None,
    ingestion_input_dir: str | None = None,
    stages: list[str] | None = None,
    document_names: list[str] | None = None,
    document_names_dir: str | None = None,
    manual: bool = False,
    model_class: str | None = None,
    only_unannotated: bool = False,
    only_unextracted: bool = False,
    context_instructions: str | None = None,
    update_mode: bool = False,
    replaces: str | None = None,
    parallel_docs: int = 5,
    on_partial_failure: Literal[
        "abort", "continue", "warn"
    ] = "warn",
    tabular_extensions: set[str] | None = None,
    tabular_delimiter: str | None = None,
) -> PipelineResult

Orchestrate the scinr-ingest pipeline end-to-end.

Chains Stages 0-4 in sequence, passing data between stages in memory when intermediate directory parameters are omitted. Tabular files (.csv, .xlsx, .xls) found in input_raw are automatically routed to the tabular pipeline.

Concurrency for LLM calls (llm_concurrency) and Neo4j writes (neo4j_concurrency) must be configured via :func:~scinr.newton.config.configure before calling this function. Use parallel_docs to control how many documents are processed concurrently across all stages.

Parameters:

Name	Type	Description	Default
input_raw	str | None	Folder containing raw source files (PDF, DOCX, CSV, XLSX, …) for Stage 0. Required when stages includes "preprocess" and extraction_input_dir is not given.	None
converter_output_dir	str | None	Folder where Stage 0 writes intermediate JSON files to disk. When None, intermediate files are kept in memory only.	None
extraction_input_dir	str | None	Folder where Stage 1 reads JSON input from disk, skipping Stage 0. Precedence: if provided, this takes absolute priority over document_names / document_names_dir for document discovery, regardless of which stages are requested — even for a stages=["annotation"]-only run, discovery reads the documents found under this folder rather than resolving document_names via Neo4j. Passing both is not an error; document_names/document_names_dir are then silently ignored. Do not pass extraction_input_dir together with document_names/document_names_dir unless this precedence is intended.	None
extraction_output_dir	str | None	Folder where Stage 1 writes extract-*.json output files.	None
ingestion_input_dir	str | None	Folder where Stage 2 reads extract-*.json files from disk, skipping Stages 0 and 1. Precedence: same absolute-priority rule as extraction_input_dir above (takes precedence over document_names/document_names_dir regardless of requested stages) — see that entry for the full explanation.	None
stages	list[str] | None	Ordered list of stage names to execute ("preprocess", "extraction", "ingestion", "annotation", "entity_extraction", "tabular"). Default runs full pipeline.	None
document_names	list[str] | None	Explicit list of Neo4j document_name values for Stage 3/4 runs. Ignored (silently) if extraction_input_dir or ingestion_input_dir is also provided — see the precedence note on those two parameters.	None
document_names_dir	str | None	Directory of extract-*.json files to extract document names from. Same silent-ignore precedence rule as document_names above.	None
manual	bool	If True, Stage 3 manual annotation assigns model_class without LLM calls.	False
model_class	str | None	CamelCase Pydantic model class name for manual annotation.	None
only_unannotated	bool	Skip nodes that already have an annotation decision.	False
only_unextracted	bool	Skip nodes that already have extracted entities.	False
context_instructions	str | None	Custom instructions injected into converter and annotation prompts.	None
update_mode	bool	If True, Stage 2 replaces latest document version in Neo4j without incrementing version.	False
replaces	str | None	document_name of existing document superseded by newly ingested document.	None
parallel_docs	int	Maximum number of documents processed concurrently (default: 1).	5
on_partial_failure	Literal['abort', 'continue', 'warn']	Control behavior when a stage fails ("abort", "continue", or "warn"). The pipeline never stops processing OTHER documents because of this flag: every document in the batch is always dispatched to the per-document-unit engine and runs independently of its siblings, regardless of on_partial_failure's value (outside the tabular short-circuit below, which still fails fast exactly as before). Within a single document's own remaining stages, the effect of on_partial_failure depends on which stage failed: "preprocess" / "extraction" / "ingestion": a failure in any of these three is a total failure for that document — no valid artifact was produced for the next stage to operate on. These always stop that document from advancing through its remaining stages, regardless of on_partial_failure (there is nothing valid to continue with). "annotation" / "entity_extraction": these operate per-node, so a stage reporting nodes_failed > 0 for a document is only a partial failure — the document itself is still valid and can proceed to its next requested stage. This only stops that document's advancement when on_partial_failure is "abort" (the default, preserving the historical per-unit "soft-abort" behavior). With "continue" or "warn", the document keeps advancing to its next requested stage even though some nodes failed in the previous one. "warn" behaves like "continue" (the document keeps advancing) but additionally logs at two levels: Immediately, a per-document warning is emitted the moment that specific document decides to keep advancing despite a partial failure in annotation or entity_extraction — naming the document, the stage, the failed-node count, and the concrete error(s) reported for it. At the end of the batch, the pre-existing aggregated per-stage warning still fires whenever a stage reports one or more failed documents overall (only the total failed-document count for that stage, not per-document detail). Both warnings coexist in "warn" mode; "continue" mode stays completely silent.	'warn'
tabular_extensions	set[str] | None	File extensions to process via tabular pipeline (default: .csv, .xlsx, .xls).	None
tabular_delimiter	str | None	Delimiter character for CSV tabular files.	None

Returns:

Type	Description
PipelineResult	PipelineResult containing stage metrics, execution flags, and duration.

Raises:

Type	Description
ConfigurationError	If Neo4j or LLM configuration is missing.
PreconditionError	If invalid parameters or mutually exclusive options are supplied.
ExtractionError	If entity extraction fails.
IngestionError	If Neo4j graph write fails.
ValueError	If any parameter combination is invalid (see validation section).
FileNotFoundError	If a required directory does not exist.