Loading data

Every indexing pipeline starts with a stream of nodes. You fill that stream from files, web pages, Kafka, Fluvio, Parquet or your own data, with a loader and Pipeline::from_loader or with any stream and Pipeline::from_stream.

Files

let languages = [SupportedLanguages::Rust, SupportedLanguages::Python];
let extensions: Vec<&str> = languages
    .iter()
    .flat_map(|language| language.file_extensions().iter().copied())
    .collect();

// Extensions without the leading dot; each call adds to the list
let loader = FileLoader::new("src")
    .with_extensions(&extensions)
    .with_extensions(&["md"]);

Pipeline::from_loader(loader)
    .then_chunk(ChunkText::from_max_characters(2000))
    .then_store_with(MemoryStorage::default())
    .run()
    .await?;

Source: snippets/src/indexing/loading.rs#L17-L32

FileLoader::new takes a directory or a single file. Its walk uses the ignore crate’s defaults. It skips hidden files and files excluded by .ignore files, and by .gitignore rules when the root is inside a git repository. with_extensions keeps only files with those extensions. Without any call the loader yields every file.

SupportedLanguages::file_extensions returns the extensions tree-sitter knows for a language. It needs the tree-sitter feature flag. FileLoader itself needs none.

Each file becomes one TextNode:

chunk holds the file content, with invalid UTF-8 replaced.
path is relative when the root is relative. The node id hashes path and chunk, so changing the root from relative to absolute misses the node cache.
A file that cannot be read becomes an error in the stream. The first error ends run unless you filter errors, see Errors.

Nodes in memory

let mut faq = TextNode::new("A loader turns a data source into a stream of nodes.");
faq.path = "faq/loading.md".into();
faq.metadata.insert("Source", "faq");

let nodes = vec![
    faq,
    TextNode::new("An indexing pipeline does nothing until you call run."),
];

let storage = MemoryStorage::default();
Pipeline::from_stream(nodes)
    .then_store_with(storage.clone())
    .run()
    .await?;

println!("Stored {} nodes", storage.get_all_values().await.len());

Source: snippets/src/indexing/loading.rs#L40-L55

Pipeline::from_stream takes anything that converts into an IndexingStream. That includes a Vec<TextNode>, a boxed Stream of results and a Tokio mpsc receiver.

A channel lets another task produce nodes while the pipeline runs:

let (sender, receiver) = tokio::sync::mpsc::channel(100);

tokio::spawn(async move {
    for line in ["first message", "second message"] {
        if sender.send(Ok(TextNode::new(line))).await.is_err() {
            break;
        }
    }
    // Dropping the sender ends the stream, and with it the run
});

Pipeline::from_stream(receiver)
    .then_store_with(MemoryStorage::default())
    .run()
    .await?;

Source: snippets/src/indexing/loading.rs#L63-L77

The stream ends when every sender is dropped. Until then, run keeps waiting for nodes.

Files from a tool executor

let executor = LocalExecutor::new(".");

let files = executor
    .stream_files(Path::new("src"), Some(vec!["rs".to_string()]))
    .await?;

Pipeline::from_stream(files)
    .then_chunk(ChunkText::from_max_characters(2000))
    .then_store_with(MemoryStorage::default())
    .run()
    .await?;

Source: snippets/src/indexing/loading.rs#L85-L95

ToolExecutor::stream_files returns the files below a path as an IndexingStream. LocalExecutor uses path as given and does not resolve it against the executor’s working directory. LocalExecutor needs the swiftide-agents feature flag.

Web pages

indexing::Pipeline::from_loader(ScrapingLoader::from_spider(
    Website::new("https://www.bosun.ai/")
        .with_limit(1)
        .to_owned(),
))
.then(HtmlToMarkdownTransformer::default())
.then_chunk(ChunkMarkdown::from_chunk_range(20..2048))
.log_all()
.then_store_with(MemoryStorage::default())
.run()
.await?;

Source: examples/scraping_index_to_markdown.rs#L26-L36

ScrapingLoader crawls with the spider crate and needs the scraping feature flag. ScrapingLoader::from_spider takes a configured Website, here limited to one page. To configure it, add spider at the version swiftide uses (2.52), or the Website types do not match.

Each node carries the raw HTML of one page, with the page URL as its path. HtmlToMarkdownTransformer converts the HTML to markdown and drops script and style tags.

Spider Cloud

Since 0.33, ScrapingLoader crawls through Spider Cloud when SPIDER_CLOUD_API_KEY is set and not empty. It reads these environment variables once per process:

Variable	Effect
`SPIDER_CLOUD_API_KEY`	Turns Spider Cloud on
`SPIDER_CLOUD_MODE`	`proxy`, `api`, `unblocker` or `fallback`. Anything else selects smart mode, which uses the unblocker only against bot protection.
`SPIDER_CLOUD_API_URL`	Optional API endpoint

Spider Cloud cannot reach hosts that only your machine can route to. A start URL on localhost, a .localhost or .local name, a loopback address, or a private or link-local IPv4 address is crawled directly.

Kafka and Fluvio

static LOADER_TOPIC: &str = "loader";
static STORAGE_TOPIC: &str = "storage";

let mut client_config = ClientConfig::new();
client_config.set("bootstrap.servers", "localhost:9092");
client_config.set("group.id", "group_id");
client_config.set("auto.offset.reset", "earliest");

let loader = Kafka::builder()
    .client_config(client_config.clone())
    .topic(LOADER_TOPIC)
    .build()
    .unwrap();

let storage = Kafka::builder()
    .client_config(client_config)
    .topic(STORAGE_TOPIC)
    .create_topic_if_not_exists(true)
    .batch_size(2usize)
    .build()
    .unwrap();

Source: examples/kafka.rs#L27-L47

indexing::Pipeline::from_loader(loader)
    .then_in_batch(Embed::new(FastEmbed::try_default().unwrap()).with_batch_size(10))
    .then_store_with(storage)
    .run()
    .await?;

Source: examples/kafka.rs#L51-L55

The pipelines embed with FastEmbed, which needs the fastembed feature flag. Kafka needs the kafka feature flag, which builds librdkafka and needs cmake and a C toolchain. It takes an rdkafka ClientConfig, re-exported as integrations::kafka::ClientConfig. As a loader it subscribes to topic and turns each message payload into a node. Messages without a UTF-8 payload are skipped. The node metadata gets kafka_topic, kafka_partition and kafka_offset, plus kafka_timestamp and kafka_key when the message has them. Kafka is also a storage, here writing to a second topic.

static TOPIC_NAME: &str = "hello-rust";
static PARTITION_NUM: u32 = 0;

let loader = Fluvio::builder()
    .consumer_config_ext(
        ConsumerConfigExt::builder()
            .topic(TOPIC_NAME)
            .partition(PARTITION_NUM)
            .offset_start(fluvio::Offset::from_end(1))
            .build()
            .unwrap(),
    )
    .build()
    .unwrap();

Source: examples/fluvio.rs#L36-L49

Fluvio needs the fluvio feature flag and is a loader only. It takes a ConsumerConfigExt with the topic, partition and start offset. Offset comes from the fluvio crate, which you add at swiftide’s version (0.50). Each record becomes a node with its key in the fluvio_key metadata field, which is null when the record has no key.

A Kafka pipeline waits for new messages, so run does not return on its own. For both loaders, a receive error ends the run.

Parquet

let loader = Parquet::builder()
    .path("data.parquet")
    .column_name("chunk")
    .build()?;

Pipeline::from_loader(loader)
    .then_chunk(ChunkText::from_max_characters(2000))
    .then_store_with(MemoryStorage::default())
    .run()
    .await?;

Source: snippets/src/indexing/loading.rs#L103-L112

Parquet needs the parquet feature flag. batch_size sets the rows read per batch and defaults to 1024. The loader reads one column and turns every non-null value into a node. The column must hold strings. A column of another type yields an error, and a missing column panics. Batches are flattened unordered, so node order can differ from row order. Each load prints the column descriptors to stderr.

Custom loaders

When no loader fits, implement Loader. The docs-qa showcase loads markdown and llms-full.txt sources as one node per page:

/// Yields one [`TextNode`] per documentation page, with the page title as metadata.
#[derive(Debug, Clone)]
pub struct DocsLoader {
    sources: Vec<DocsSource>,
}

impl DocsLoader {
    pub fn new(sources: Vec<DocsSource>) -> Self {
        Self { sources }
    }
}

impl Loader for DocsLoader {
    type Output = String;

    fn into_stream(self) -> IndexingStream<String> {
        stream::iter(self.sources)
            .then(DocsSource::pages)
            .map_ok(|pages| stream::iter(pages).map(|page| Ok(page_node(page))))
            .try_flatten()
            .boxed()
            .into()
    }

    // Only needed when the loader is used as a `Box<dyn Loader>`.
    fn into_stream_boxed(self: Box<Self>) -> IndexingStream<String> {
        self.into_stream()
    }
}

fn page_node(page: Page) -> TextNode {
    let mut node = TextNode::new(page.markdown);
    node.path = page.origin.into();
    node.metadata.insert(PAGE_TITLE, page.title.as_str());
    node
}

Source: examples/docs-qa/src/loader.rs#L74-L109

type Output is the chunk type of the nodes. Built-in transformers expect String.
into_stream is synchronous and consumes the loader. Put slow work inside the stream, like the page downloads this loader starts through DocsSource::pages, so it runs when the pipeline polls it.
A failing source becomes an Err item, not a panic.
into_stream_boxed is only needed when you use the loader as a Box<dyn Loader>. Its default implementation panics.
Loader requires Send, Sync and DynClone, which any Clone type implements.

Gotchas

Pipeline::from_loader calls into_stream right away, so work starts when you build the pipeline, not when you call run. Kafka subscribes to its topic. Fluvio connects. Parquet opens the file. ScrapingLoader starts crawling, and panics outside a Tokio runtime.
Fluvio and Parquet use tokio::task::block_in_place, which panics on a current-thread runtime. #[tokio::main] uses a multi-thread runtime by default; #[tokio::test] does not, so use #[tokio::test(flavor = "multi_thread")].
Kafka, Fluvio and Parquet panic when they cannot create their consumer or open their file.

Feature flags

Install the feature flags you need, for example:

cargo add swiftide@0.34 --features tree-sitter,scraping,kafka,fluvio,parquet,fastembed

Feature flags lists them all.

Reference

Chunking