Using IngestionPipeline for content not originating from the file system

[f2bo]

I'm beginning to look at Microsoft.Extensions.DataIngestion pipelines. As a test, I considered using an IngestionPipeline to ingest content stored in a CMS SQL database and create a vector store for use with RAG. However, I'm unclear on how to implement it when the data to be ingested is stored in a database.

Currently, both overloads of the ProcessAsync method require file system objects.

extensions/src/Libraries/Microsoft.Extensions.DataIngestion/IngestionPipeline.cs

Lines 80 to 81 in 15ffd76

	public async IAsyncEnumerable<IngestionResult> ProcessAsync(DirectoryInfo directory, string searchPattern = "*.*",
	SearchOption searchOption = SearchOption.TopDirectoryOnly, [EnumeratorCancellation] CancellationToken cancellationToken = default)

and

extensions/src/Libraries/Microsoft.Extensions.DataIngestion/IngestionPipeline.cs

Lines 107 to 108 in 15ffd76

	public async IAsyncEnumerable<IngestionResult> ProcessAsync(IEnumerable<FileInfo> files, [EnumeratorCancellation] CancellationToken cancellationToken = default)
	{

Perhaps I misunderstand its purpose or how it's meant to be used, but it would appear that it can only ingest data originating from files. Is that the case?

[luisquintanilla]

cc: @adamsitnik

[adamsitnik]

@f2bo Big thanks for sharing the feedback. Let me prototype a design proposal and get back to you really quick.

[f2bo]

@adamsitnik Thanks for taking a look.

If it helps, after I wrote my previous message, I played a bit with the code and came up with the following changes that worked for me.

IngestionPipeline processes a list of documents as specified by their keys or IDs, without any knowledge of their origin, which is exclusively handled by the IngestionDocumentReader.

/// <summary>
/// Represents a pipeline for ingesting data from documents and processing it into chunks.
/// </summary>
/// <typeparam name="TChunk">The type of the chunk content.</typeparam>
/// <typeparam name="TKey">The type of the document key.</typeparam>
public sealed class IngestionPipeline<TChunk, TKey> : IDisposable
{
    ...

    /// <summary>
    /// Processes multiple documents matching an optional predicate or all available documents.
    /// </summary>
    /// <param name="documentFilter">An optional expression to filter which documents IDs to process.</param>
    /// <param name="cancellationToken">The cancellation token for the operation.</param>
    /// <returns>A task representing the asynchronous operation.</returns>
    public async IAsyncEnumerable<IngestionResult> ProcessAsync(Func<TKey, bool>? documentFilter = null, [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        ...
    }

    /// <summary>
    /// Processes multiple documents for a list of specific IDs.
    /// </summary>
    /// <param name="documentIds">The collection of documents IDs to process.</param>
    /// <param name="cancellationToken">The cancellation token for the operation.</param>
    /// <returns>A task representing the asynchronous operation.</returns>
    public async IAsyncEnumerable<IngestionResult> ProcessAsync(IEnumerable<TKey> documentIds, [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
    }

    ...
}

IngestionDocumentReader is the one that is aware of the content's location. It receives a one or more document keys and converts them to IngestionDocuments.

/// <summary>
/// Reads source content and converts it to an <see cref="IngestionDocument"/>.
/// </summary>
/// <typeparam name="TKey">The type of the document key.</typeparam>
public abstract class IngestionDocumentReader<TKey>
{
    /// <summary>
    /// Reads a single document for a specific ID.
    /// </summary>
    public abstract Task<IngestionDocument?> ReadAsync(TKey documentId, CancellationToken cancellationToken = default);

    /// <summary>
    /// Reads multiple documents for a list of specific IDs.
    /// </summary>
    public abstract IAsyncEnumerable<IngestionDocument> ReadAsync(IEnumerable<TKey> documentIds, CancellationToken cancellationToken = default);

    /// <summary>
    /// Reads multiple documents matching an optional predicate or all available documents.
    /// </summary>
    public abstract IAsyncEnumerable<IngestionDocument> ReadAsync(Func<TKey, bool>? documentFilter = null, CancellationToken cancellationToken = default);
}

For my scenario, where the document reader I implemented retrieves content from a database, the document keys correspond to database record keys. I haven't tested it, but I would imagine that a reader that works with files could use filenames as keys.

Let me know if you want me to include my implementation.

[adamsitnik]

@f2bo Big thanks for your input.

The IngestionDocumentReader class supports reading documents from files and streams. Its responsibility is to take given input, parse it using some parser (MarkItDown, LlamaParse, Azure Document Intelligence) and then map from the parser format to IngestionDocument (class for representing format-agnostic document containers). In theory it could read something from the DB as well, but then it would be reponsible for parsing it and mapping to IngestionDocument type.

In the initial design of the IngestionPipeline, we had two ProcessAsync methods that both accepted only input from the file system:

• ProcessAsync(DirectoryInfo, ...),
• ProcessAsync(IEnumerable<FileInfo>, ...).

The main reason why we have not added similar methods for streams was that using a stream as input usually requires handling its lifecycle.
And accepting an IEnumerable<Stream> would complicate the API, as the caller would need to ensure they are truly lazy.
By lazy I mean created when needed, as if the caller simply opened too many streams at once they could run out of resources.

To address this issue I propose (#7087) we add two ProcessAsync methods that accept a single Stream or a buffer.
Since these methods handle only one document at a time, the caller is responsible for managing the stream lifecycle.
And the users won't be forced to use streams, as they can use the buffer overload that takes ReadOnlyMemory<byte>.
And we don't need to worry about partial success handling, as the operation either succeeds or fails for that single document.

@f2bo Would the proposed API solve your problem? The way I imagine its usage:

using IngestionPipeline<string> pipeline = new(reader, chunker, writer);
using SomeDbRepository dbRepository = new("someConnectionString");

await foreach (var document in dbRepository.GetDocumentsAsync())
{
    await pipeline.ProcessAsync(document.Blob, document.Id, document.MediaType);
}

[SamMonoRT]

cc @roji fyi

[f2bo]

The IngestionDocumentReader class supports reading documents from files and streams.

This is what I believe unnecessarily constrains the design. Given a list of documents IDs, each reader should know how to interpret these IDs and retrieve their content. Whether it does this by reading from a file, a stream, a database, or an API request should only be an implementation detail. Publicly, the reader only knows about document IDs, which are specific to each reader.

In theory it could read something from the DB as well, but then it would be reponsible for parsing it and mapping to IngestionDocument type.

Correct. This is how I handled it. The reader knows the database schema and is able to create an IngestionDocument from each requested record. It's no different from the Markdig document reader in this repository, which parses a PDF documents and creates IngestionDocument's.

The main reason why we have not added similar methods for streams was that using a stream as input usually requires handling its lifecycle.

This shouldn't be necessary if it's the IngestionDocumentReader that handles it instead of the IngestionPipeline . The IngestionPipeline should be completely agnostic of the content's origin and only deals with document IDs, which can be database record IDs, filenames, FileInfo, Uri, etc...

@f2bo Would the proposed API solve your problem?

I imagine it would, but I'll turn it around. There's no need for additional methods in the IngestionPipeline or handling lifecycles if each reader has the knowledge about its specific content type and knows how to retrieve and parse it.

For example, using a document reader that reads from a database using integer document IDs.

using IngestionPipeline<string, int> pipeline = new(DatabaseDocumentReader<int>, chunker, writer);

List<int> documentIds = [1000, 1200, 1201, 1202, 1234, 1315];
await foreach (IngestionResult result in pipeline.ProcessAsync(documentIds, cancellationToken))
{
    logger.LogInformation("Completed processing '{id}'. Succeeded: '{succeeded}'.", result.DocumentId, result.Succeeded);
}

Or using a document reader that reads from disk files using filenames as IDs.

using IngestionPipeline<string, string> pipeline = new(new PdfFileDocumentReader<string>(), chunker, writer);

List<string> filenames = Directory.GetFiles(somepath, "*.pdf");
await foreach (IngestionResult result in pipeline.ProcessAsync(filenames, cancellationToken))
{
    logger.LogInformation("Completed processing '{id}'. Succeeded: '{succeeded}'.", result.DocumentId, result.Succeeded);
}

Or using a document reader that makes HTTP requests using URLs as IDs.

using IngestionPipeline<string, Uri> pipeline = new(new HttpRequestDocumentReader<Uri>(), chunker, writer);

List<Uri> uris = [
    new Uri("https://xyz.com/page1"), 
    new Uri("https://xyz.com/page2"), 
    new Uri("https://xyz.com/page3"), 
];

await foreach (IngestionResult result in pipeline.ProcessAsync(uris, cancellationToken))
{
    logger.LogInformation("Completed processing '{id}'. Succeeded: '{succeeded}'.", result.DocumentId, result.Succeeded);
}

All the usages above are calling the same method in the IngestionPipeline, which only needs this single method.

public async IAsyncEnumerable<IngestionResult> ProcessAsync(IEnumerable<TKey> documentIds, [EnumeratorCancellation] CancellationToken cancellationToken = default)

[adamsitnik]

Big thanks for your feedback @f2bo!

When I was designing the initial version, the most important scenario was allowing the users to ingest existing files (and whole directories). To allow for other inputs, we added the overload that takes the Stream.

When it comes to reading from a DB using any key, the current Reader and Pipeline design does not allow for that (as you can see). I've one idea that I'll try to implement (introducing and interface where TSource would be a generic param and existing reader types could implement multiple TSoruce inputs), but I already know that just exposing TKey (or rather TSource) won't be enough as some parsers require additional data like content type.

In the meantime you can use the rest of our building blocks: chunkers, writer and enrichers.

Sample:

public class DbPipeline<TKey>
{
    private readonly Func<TKey, Task<IngestionDocument>> _dbReader;
    private readonly IngestionChunker<T> _chunker;
    private readonly IngestionChunkWriter<T> _writer;

    public IList<IngestionDocumentProcessor> DocumentProcessors { get; } = [];
    public IList<IngestionChunkProcessor<T>> ChunkProcessors { get; } = [];

    public DbPipeline(Func<TKey, Task<IngestionDocument>> dbReader, IngestionChunker<T> chunker, IngestionChunkWriter<T> writer)
    {
        _dbReader = dbReader;
        _chunker = chunker;
        _writer = writer;
    }

    public async Task ProcessAsync(TKey key, CancellationToken cancellationToken = default)
    {
        IngestionDocument document = await _dbReader(key).ConfigureAwait(false);

        foreach (IngestionDocumentProcessor processor in DocumentProcessors)
        {
            document = await processor.ProcessAsync(document, cancellationToken).ConfigureAwait(false);
        }

        IAsyncEnumerable<IngestionChunk<T>> chunks = _chunker.ProcessAsync(document, cancellationToken);
        foreach (var processor in ChunkProcessors)
        {
            chunks = processor.ProcessAsync(chunks, cancellationToken);
        }

        await _writer.WriteAsync(chunks, cancellationToken).ConfigureAwait(false);
    }
}

Again, thanks a lot for giving it a try and sharing the feedback. It's perfect moment for me as the project is still in preview (we can introduce breaking changes) and also in active development.