Status callback for import methods

[akordowski]

As I understand there is no way to determine if the ingestion pipeline is running in process or distributed. The only way I see is to call the GetDocumentStatusAsync or IsDocumentReadyAsync methods periodically.

How to provide a optional status callback parameter to the import methods (eg. Func<DataPipelineStatus>? statusCallback = null) or a OrchestrationType property to the IKernelMemory interface?

[akordowski]

@dluc Currently I use the bellow method to get the document status. But since the data ingestion process has the status anyway, it would be more efficient if we could pass a callback directly to the kernelMemory.ImportDocumentAsync() method, instead of pulling the status by delay.

Any thoughts on this?

public static async Task CheckDocumentStatusAsync(
    this IKernelMemory kernelMemory,
    string documentId,
    string? index = null,
    Func<int, int, DocumentStatus, CancellationToken, Task>? progressCallback = null,
    int millisecondsDelay = 1000,
    CancellationToken cancellationToken = default)
{
    while (true)
    {
        var pipelineStatus = await kernelMemory
            .GetDocumentStatusAsync(documentId, index, cancellationToken)
            .ConfigureAwait(false);

        var step = 0;
        var stepsTotal = 0;
        var status = DocumentStatus.None;

        if (pipelineStatus is not null)
        {
            step = pipelineStatus.CompletedSteps.Count;
            stepsTotal = pipelineStatus.Steps.Count;

            if (step < stepsTotal)
            {
                step++;
            }

            if (pipelineStatus.Empty)
            {
                status = DocumentStatus.Empty;
            }
            else if (pipelineStatus.Failed)
            {
                status = DocumentStatus.Failed;
            }
            else if (pipelineStatus.Completed)
            {
                status = DocumentStatus.Completed;
            }
            else
            {
                status = DocumentStatus.InProgress;
            }
        }

        if (progressCallback is not null)
        {
            await progressCallback(step, stepsTotal, status, cancellationToken).ConfigureAwait(false);
        }

        if (status == DocumentStatus.InProgress)
        {
            await Task.Delay(millisecondsDelay, cancellationToken).ConfigureAwait(false);
        }
        else
        {
            break;
        }
    }
}
´´´

[dluc]

Hi @akordowski, I appreciate that this is a recurring scenario, in fact it comes up from time to time.

KM API (IKernelMemory interface) is available over 3 implementations, that we aim to keep 100% compatible:

• Web API, aka MemoryWebClient
• Asynchronous instance, aka MemoryService
• Synchronous instance, aka MemoryServerless

If we introduce a callback into IKernelMemory.ImportDocumentAsync, then it would have to be implemented also in the web service to inform MemoryWebClient. Callbacks over HTTP can be unreliable and open new security concerns, so we don't plan to implement this in the KM interface.

On the other hand, you can easily introduce a custom callback using a custom handler. The default ingestion pipeline goes through these steps:

• extract text, chunk, generate embeddings, store records.

The list can be overridden using the steps parameter, so you could introduce a final step:

• extract text, chunk, generate embeddings, store records, mycallback.

[akordowski]

The list can be overridden using the steps parameter, so you could introduce a final step:
extract text, chunk, generate embeddings, store records, mycallback.

As I understand this approach would only have one callback at the end of the process.

What I have in mind is rather a progress callback everytime the status of the pipeline changes.

As showed in my example this progress callback would be an optional Func<int, int, DocumentStatus, CancellationToken, Task>? progressCallback = null method parameter. This function would be only invoked when available (progressCallback?.Invoke()) and provide values like cuurentStep, totalSteps and pipelineStatus enum (see #709). I haven't inspected the KM code yet, but this should be doable, as the same data is kinda available by callling the GetDocumentStatusAsync() method.

If we introduce a callback into IKernelMemory.ImportDocumentAsync, then it would have to be implemented also in the web service to inform MemoryWebClient.

I don't see why this would necessarily have to be implemented also in the Web API, as this would be optional and just a helper so we can get the progress without calling the GetDocumentStatusAsync() method.

I hope I could explain my idea and intentions better.

[dluc]

As I understand this approach would only have one callback at the end of the process.

The pipeline is fully customizable, the list of steps is... well... a list :-). To receive a call back after each action, the list would look like: extract text, callback 1, chunk, callback 2, generate embeddings, callback 3, store records, callback 4.

[akordowski]

It might be a workaround, but it is far from being a good solution. Are there any technical objections to consider the proposed solution?

[dluc]

How would you implement the call back in MemoryWebClient?

[akordowski]

I thought the whole time you were talking about an REST API. Now I see there is a MemoryWebClient class.

As I understand the class we could provide an implementaion similar to the example I posted above. The web client could periodically call the API endpoint equivalent of the GetDocumentStatusAsync() method, only if the progressCallback function is provided. In this way it would not have any impact on the currrent implementaion. The delay we could provide as a constructor parameter.

On the second thought this could lead to long time blocking of the ImportDocumentAsync() method. I'm not so deep into the code base, but maybe we could fire and forget the async task for the progressCallback in order not to block the ImportDocumentAsync() execution? Maybe you can say more about this.