Implementation:Run llama Llama index IngestionPipeline Persist

Knowledge Sources	LlamaIndex
Domains	Data_Ingestion, RAG, Data_Management
Last Updated	2026-02-11 00:00 GMT

Overview

The IngestionPipeline.persist() and IngestionPipeline.load() methods serialize and deserialize the pipeline's cache and docstore to/from disk, enabling resumable and incremental ingestion across process restarts.

Description

persist() writes the pipeline's IngestionCache and BaseDocumentStore (if present) to JSON files in the specified directory. load() reads these files back and restores the pipeline's internal state. Together, they enable a workflow where the pipeline remembers which documents have been processed and what transformation results have been cached.

The underlying IngestionCache class provides its own persistence interface (persist() and from_persist_path()) as well as direct put(), get(), and clear() methods for programmatic cache management.

Usage

Call persist() after each successful pipeline run, and load() before the next run. Use a consistent persist_dir path across runs.

Code Reference

Source Location

Repository: llama_index
File (pipeline): llama-index-core/llama_index/core/ingestion/pipeline.py
Lines (pipeline): L317-364
File (cache): llama-index-core/llama_index/core/ingestion/cache.py
Lines (cache): L17-75

Signature: persist()

def persist(
    self,
    persist_dir: str = "./pipeline_storage",
    fs: Optional[AbstractFileSystem] = None,
    cache_name: str = DEFAULT_CACHE_NAME,
    docstore_name: str = DOCSTORE_FNAME,
) -> None:

Signature: load()

def load(
    self,
    persist_dir: str = "./pipeline_storage",
    fs: Optional[AbstractFileSystem] = None,
    cache_name: str = DEFAULT_CACHE_NAME,
    docstore_name: str = DOCSTORE_FNAME,
) -> None:

IngestionCache Methods

class IngestionCache(BaseModel):
    def put(self, key: str, nodes: List[BaseNode], collection: str = ...) -> None:
        """Store transformation results in the cache."""

    def get(self, key: str, collection: str = ...) -> Optional[List[BaseNode]]:
        """Retrieve cached transformation results."""

    def clear(self) -> None:
        """Clear all cached entries."""

    def persist(self, persist_path: str, fs: Optional[AbstractFileSystem] = None) -> None:
        """Serialize cache to disk."""

    @classmethod
    def from_persist_path(cls, persist_path: str, fs: Optional[AbstractFileSystem] = None) -> "IngestionCache":
        """Deserialize cache from disk."""

Import

from llama_index.core.ingestion import IngestionPipeline

# Usage (method calls on instance)
pipeline.persist("./storage")
pipeline.load("./storage")

I/O Contract

Inputs (persist)

Name	Type	Required	Description
persist_dir	str	No (default: "./pipeline_storage")	Directory path for storing serialized state files
fs	Optional[AbstractFileSystem]	No	fsspec filesystem abstraction for remote storage support
cache_name	str	No (default: DEFAULT_CACHE_NAME)	Filename for the serialized cache
docstore_name	str	No (default: DOCSTORE_FNAME)	Filename for the serialized docstore

Inputs (load)

Name	Type	Required	Description
persist_dir	str	No (default: "./pipeline_storage")	Directory path containing previously persisted state files
fs	Optional[AbstractFileSystem]	No	fsspec filesystem abstraction for remote storage support
cache_name	str	No (default: DEFAULT_CACHE_NAME)	Filename of the serialized cache to load
docstore_name	str	No (default: DOCSTORE_FNAME)	Filename of the serialized docstore to load

Outputs

Name	Type	Description
return (persist)	None	State is written to disk as side effect
return (load)	None	Pipeline cache and docstore are restored from disk as side effect

Usage Examples

Persist After Ingestion

from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.storage.docstore import SimpleDocumentStore

pipeline = IngestionPipeline(
    transformations=[SentenceSplitter()],
    docstore=SimpleDocumentStore(),
)

# First run - processes all documents
nodes = pipeline.run(documents=documents)

# Save state to disk
pipeline.persist("./pipeline_storage")

Load and Resume

from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.storage.docstore import SimpleDocumentStore

pipeline = IngestionPipeline(
    transformations=[SentenceSplitter()],
    docstore=SimpleDocumentStore(),
)

# Restore state from previous run
pipeline.load("./pipeline_storage")

# Second run - only processes new or changed documents
nodes = pipeline.run(documents=updated_documents)

# Save updated state
pipeline.persist("./pipeline_storage")

Related Pages

Implements Principle

Principle:Run_llama_Llama_index_Pipeline_State_Persistence

Requires Environment

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment