Struct Pipeline

pub struct Pipeline { /* private fields */ }

A pipeline for indexing files, adding metadata, chunking, transforming, embedding, and then storing them.

The Pipeline struct orchestrates the entire file indexing process. It is designed to be flexible and performant, allowing for various stages of data transformation and storage to be configured and executed asynchronously.

Fields

• stream - The stream of Node items to be processed.
• storage - Optional storage backend where the processed nodes will be stored.
• concurrency - The level of concurrency for processing nodes.

Implementations

impl Pipeline

pub fn from_loader(loader: impl Loader + 'static) -> Self

Creates a Pipeline from a given loader.

Arguments

• loader - A loader that implements the Loader trait.

Returns

An instance of Pipeline initialized with the provided loader.

pub fn with_default_llm_client( self, client: impl SimplePrompt + 'static, ) -> Self

Sets the default LLM client to be used for LLM prompts for all transformers in the pipeline.

pub fn from_stream(stream: impl Into<IndexingStream>) -> Self

Creates a Pipeline from a given stream.

Arguments

• stream - An IndexingStream containing the nodes to be processed.

Returns

An instance of Pipeline initialized with the provided stream.

pub fn with_concurrency(self, concurrency: usize) -> Self

Sets the concurrency level for the pipeline. By default the concurrency is set to the number of cpus.

Arguments

• concurrency - The desired level of concurrency.

Returns

An instance of Pipeline with the updated concurrency level.

pub fn with_embed_mode(self, embed_mode: EmbedMode) -> Self

Sets the embed mode for the pipeline. The embed mode controls what (combination) fields of a Node be embedded with a vector when transforming with crate::transformers::Embed

See also swiftide_core::indexing::EmbedMode.

Arguments

• embed_mode - The desired embed mode.

Returns

An instance of Pipeline with the updated embed mode.

pub fn filter_cached(self, cache: impl NodeCache + 'static) -> Self

Filters out cached nodes using the provided cache.

Arguments

• cache - A cache that implements the NodeCache trait.

Returns

An instance of Pipeline with the updated stream that filters out cached nodes.

pub fn then( self, transformer: impl Transformer + WithIndexingDefaults + 'static, ) -> Self

Adds a transformer to the pipeline.

Closures can also be provided as transformers.

Arguments

• transformer - A transformer that implements the Transformer trait.

Returns

An instance of Pipeline with the updated stream that applies the transformer to each node.

pub fn then_in_batch( self, transformer: impl BatchableTransformer + WithBatchIndexingDefaults + 'static, ) -> Self

Adds a batch transformer to the pipeline.

If the transformer has a batch size set, the batch size from the transformer is used, otherwise the pipeline default batch size ([DEFAULT_BATCH_SIZE]).

Arguments

• transformer - A transformer that implements the BatchableTransformer trait.

Returns

An instance of Pipeline with the updated stream that applies the batch transformer to each batch of nodes.

pub fn then_chunk(self, chunker: impl ChunkerTransformer + 'static) -> Self

Adds a chunker transformer to the pipeline.

Arguments

• chunker - A transformer that implements the ChunkerTransformer trait.

Returns

An instance of Pipeline with the updated stream that applies the chunker transformer to each node.

pub fn then_store_with(self, storage: impl Persist + 'static) -> Self

Persists indexing nodes using the provided storage backend.

Arguments

• storage - A storage backend that implements the Storage trait.

Returns

An instance of Pipeline with the configured storage backend.

Panics

Panics if batch size turns out to be not set and batch storage is still invoked. Pipeline only invokes batch storing if the batch size is set, so should be alright.

pub fn split_by<P>(self, predicate: P) -> (Self, Self)

where P: Fn(&Result<Node>) -> bool + Send + Sync + 'static,

Splits the stream into two streams based on a predicate.

Note that this is not lazy. It will start consuming the stream immediately and send each item to the left or right stream based on the predicate.

The other streams have a buffer, but should be started as soon as possible. The channels of the resulting streams are bounded and the parent stream will panic if sending fails.

They can either be run concurrently, alternated between or merged back together.

Panics

Panics if the receiving pipelines buffers are full or unavailable.

pub fn merge(self, other: Self) -> Self

Merges two streams into one

This is useful for merging two streams that have been split using the split_by method.

The full stream can then be processed using the run method.

pub fn throttle(self, duration: impl Into<Duration>) -> Self

Throttles the stream of nodes, limiting the rate to 1 per duration.

Useful for rate limiting the indexing pipeline. Uses tokio_stream::StreamExt::throttle internally which has a granualarity of 1ms.

pub fn filter_errors(self) -> Self

pub fn filter<F>(self, filter: F) -> Self

where F: Fn(&Result<Node>) -> bool + Send + Sync + 'static,

Provide a closure to selectively filter nodes or errors

This allows you to skip specific errors or nodes, or do ad hoc inspection.

If the closure returns true, the result is kept, otherwise it is skipped.

pub fn log_all(self) -> Self

Logs all results processed by the pipeline.

This method logs all results processed by the pipeline at the DEBUG level.

pub fn log_errors(self) -> Self

Logs all errors encountered by the pipeline.

This method logs all errors encountered by the pipeline at the ERROR level.

pub fn log_nodes(self) -> Self

Logs all nodes processed by the pipeline.

This method logs all nodes processed by the pipeline at the DEBUG level.

pub async fn run(self) -> Result<()>

Runs the indexing pipeline.

This method processes the stream of nodes, applying all configured transformations and storing the results.

Returns

A Result indicating the success or failure of the pipeline execution.

Errors

Returns an error if no storage backend is configured or if any stage of the pipeline fails.

Trait Implementations

impl Default for Pipeline

fn default() -> Self

Creates a default Pipeline with an empty stream, no storage, and a concurrency level equal to the number of CPUs.