batch-indexing.mdx

sidebar_position	3

Batch Indexing

Index multiple documents efficiently with progress tracking and error handling.

Python

import asyncio
from vectorless import Engine, IndexContext, IndexOptions

async def main():
    engine = Engine(
        api_key="sk-...",
        model="gpt-4o",
    )

    # Index a directory of documents
    result = await engine.index(
        IndexContext.from_dir("./documents/")
    )

    print(f"Indexed {len(result.items)} documents")
    print(f"Failures: {len(result.failed)}")

    for item in result.items:
        print(f"  ✓ {item.name} ({item.format}) → {item.doc_id}")
        if item.metrics:
            m = item.metrics
            print(f"    Nodes: {m.nodes_processed}, "
                  f"Summaries: {m.summaries_generated}, "
                  f"Time: {m.total_time_ms}ms")

    for fail in result.failed:
        print(f"  ✗ {fail.source}: {fail.error}")

    # List all indexed documents
    docs = await engine.list()
    print(f"\nTotal indexed: {len(docs)} documents")

asyncio.run(main())

Rust

use vectorless::{Engine, EngineBuilder, IndexContext};

#[tokio::main]
async fn main() -> vectorless::Result<()> {
    let engine = EngineBuilder::new()
        .with_key("sk-...")
        .with_model("gpt-4o")
        .build()
        .await?;

    // Index a directory
    let result = engine.index(IndexContext::from_dir("./documents/")).await?;

    println!("Indexed {} documents", result.items.len());
    println!("Failures: {}", result.failed.len());

    for item in &result.items {
        println!("  ✓ {} ({:?}) → {}", item.name, item.format, item.doc_id);
    }

    // List all documents
    let docs = engine.list().await?;
    println!("Total indexed: {} documents", docs.len());

    Ok(())
}

Error Handling

Each item in the result is either successful or failed. Failures don't prevent other documents from being indexed:

result = await engine.index(IndexContext.from_paths(mixed_paths))

# Successful items
for item in result.items:
    process(item)

# Failed items — handle gracefully
for fail in result.failed:
    print(f"Failed: {fail.source} — {fail.error}")