Checkpointing pipeline execution at specific components usingpipeline.run(..., until='component_x',...)`
Resuming pipeline execution from saved states usingpipeline.run(..., from_='component_y',...)`
Dumping/loading pipeline state usingdump_state(pipeline_run_id) andload_state(state).

The state includes results from previous runs. This feature is particularly useful for:

Debugging long-running pipelines
Recovering from failures
Comparing component implementations with deterministic inputs

Type of Change

New feature
Bug fix
Breaking change
Documentation update
Project configuration change

Complexity

Complexity: high

How Has This Been Tested?

Unit tests
E2E tests
Manual tests

Checklist

The following requirements should have been met (depending on the changes in the branch):

Documentation has been updated
Unit tests have been updated
E2E tests have been updated
Examples have been updated
New files have copyright header
CLA (https://neo4j.com/developer/cla/) has been signed
CHANGELOG.md updated if appropriate

NathalieCharbel added6 commits

June 4, 2025 11:19

Serialize/deserialize component state

4a93761

Run pipeline until/Resume pipeline from

f940506

Remove in memory storage support for pipeline state

12d2e26

Add pipeline run_id and ability to save and load state from json file

8a00015

Add unit tests

5f0c892

Update changelog and docs

22722b8

NathalieCharbel requested a review froma team as acode owner

June 4, 2025 10:18

Ruff

9a57046

stellasia reviewed

Jun 9, 2025

View reviewed changes

docs/source/user_guide_pipeline.rst OutdatedShow resolvedHide resolved

src/neo4j_graphrag/experimental/pipeline/pipeline.py OutdatedShow resolvedHide resolved

tests/unit/experimental/pipeline/components.py OutdatedShow resolvedHide resolved

docs/source/user_guide_pipeline.rst OutdatedShow resolvedHide resolved

src/neo4j_graphrag/experimental/pipeline/pipeline.py OutdatedShow resolvedHide resolved

NathalieCharbel added4 commits

June 10, 2025 15:45

Remove state management for component

d1f7389

Remove resume_from and run_until and reuse existing run interface

8bc8ac0

Add dump and load to InMemoryStore

f76e5ca

Ruff

2cc5b99

stellasia reviewed

Jun 11, 2025

View reviewed changes

src/neo4j_graphrag/experimental/pipeline/stores.py OutdatedShow resolvedHide resolved

NathalieCharbel added3 commits

June 11, 2025 16:44

Add ability ro load and dump state by run_id

2a819ea

Allow orchestrator to run use a run_id from previous run

472eaf1

Refactor pipeline and validate loaded state

13254dd

stellasia reviewed

Jun 11, 2025

View reviewed changes

src/neo4j_graphrag/experimental/pipeline/stores.py Outdated

		keys_to_remove = [
		key for key in self._data.keys() if key.startswith(run_id_prefix)
		]
		for key in keys_to_remove:

Copy link

Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others.Learn more.

So here we are removing all results from a previous run with this run_id, right?

Copy link

ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others.Learn more.

yes!

stellasia reviewed

Jun 11, 2025

View reviewed changes

src/neo4j_graphrag/experimental/pipeline/pipeline.py Outdated

		@@ -140,6 +140,7 @@ def __init__(
		}
		"""
		self.missing_inputs: dict[str, list[str]] = defaultdict()
		self._current_run_id: Optional[str] = None

Copy link

Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others.Learn more.

This can not be saved in the Pipeline instance, since concurrent runs will override it.

Copy link

ContributorAuthor

NathalieCharbelJun 12, 2025•
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others.Learn more.

I will move it back todump() function. I think we should maintain creating differentrun_ids even after resuming the same pipeline and dump the state based on previous ones. we could keep track ofrun_ids of the same pipeline in the state. This should resolve the concurrency issue, right?