Parallel execution of steps and aggregating the results

[bux]

What is the best way of handling the results of parallel steps in a process (or even nested processes)?

I've added an AggregatingStep that handles multiple events and sets the results on its internal state.
Each handler then checks of all states have been and then emits an event with all the results combined for the next step in the process.

This feels clunky.

Is there a better way?

flowchart LR
    Start["Start"]
    End["End"]
    Step1["Step1"]
    Step1["Step1"] --> AggregationStep["AggregationStep"]
    Step2["Step2"]
    Step2["Step2"] --> AggregationStep["AggregationStep"]
    Step3["Step3"]
    Step3["Step3"] --> AggregationStep["AggregationStep"]
    Step4["Step4"]
    Step4["Step4"] --> AggregationStep["AggregationStep"]
    AggregationStep["AggregationStep"]
    EndStep["EndStep"]
    EndStep["EndStep"] --> End["End"]
    AggregationStep["AggregationStep"] --> EndStep["EndStep"]
    Start --> Step1["Step1"]
    Start --> Step2["Step2"]
    Start --> Step3["Step3"]
    Start --> Step4["Step4"]

Loading

Example code of an aggeration step

using Microsoft.SemanticKernel;

namespace Example;

public class AggregationStep : KernelProcessStep<DataAggregationStep.AggregateState>
{
    private AggregateState _state = new();

    public override ValueTask ActivateAsync(KernelProcessStepState<AggregateState> state)
    {
        _state = state.State!;
        return base.ActivateAsync(state);
    }

    [KernelFunction(Functions.DataPoint1WasAnalyzed)]
    public async Task DataPoint1WasAnalyzed(Kernel kernel, KernelProcessStepContext context, string response)
    {
        _state.DataPoint1AnalysisResponse = response;
        _state.DataPoint1Analyzed = true;
        await CheckIfAnalysisIsCompleteAsync(context);
    }
    
    [KernelFunction(Functions.DataPoint2WasAnalyzed)]
    public async Task DataPoint2WasAnalyzed(Kernel kernel, KernelProcessStepContext context, string response)
    {
        _state.DataPoint2AnalysisResponse = response;
        _state.DataPoint2Analyzed = true;
        await CheckIfAnalysisIsCompleteAsync(context);
    }

    [KernelFunction(Functions.DataPoint3WasAnalyzed)]
    public async Task DataPoint3WasAnalyzed(Kernel kernel, KernelProcessStepContext context, string response)
    {
        _state.DataPoint3AnalysisResponse = response;
        _state.DataPoint3Analyzed = true;
        await CheckIfAnalysisIsCompleteAsync(context);
    }
    
    [KernelFunction(Functions.DataPoint4WasAnalyzed)]
    public async Task DataPoint4WasAnalyzed(Kernel kernel, KernelProcessStepContext context, string response)
    {
        _state.DataPoint4AnalysisResponse = response;
        _state.DataPoint4Analyzed = true;
        await CheckIfAnalysisIsCompleteAsync(context);
    }

    private async Task CheckIfAnalysisIsCompleteAsync(KernelProcessStepContext context)
    {
        if (_state.DataPoint1Analyzed && 
            _state.DataPoint2Analyzed &&
            _state.DataPoint3Analyzed &&
            _state.DataPoint4Analyzed)
            await context.EmitEventAsync(OutputEvents.AnalysisComplete, AggregateResults());
    }

    
    private string AggregateResults()
    {
        return 
            string.Join(",", 
                _state.DataPoint1AnalysisResponse, 
                _state.DataPoint2AnalysisResponse,
                _state.DataPoint3AnalysisResponse,
                _state.DataPoint4AnalysisResponse
            );
    }
    
    public static class Functions
    {
        public const string DataPoint1WasAnalyzed = nameof(DataPoint1WasAnalyzed);
        public const string DataPoint2WasAnalyzed = nameof(DataPoint2WasAnalyzed);
        public const string DataPoint3WasAnalyzed = nameof(DataPoint3WasAnalyzed);
        public const string DataPoint4WasAnalyzed = nameof(DataPoint4WasAnalyzed);
    }

    public static class OutputEvents
    {
        public const string AnalysisComplete = nameof(AnalysisComplete);
    }

    public class AggregateState
    {
        public bool DataPoint1Analyzed { get; set; }
        public string DataPoint1AnalysisResponse { get; set; }
        public bool DataPoint2Analyzed { get; set; }
        public string DataPoint2AnalysisResponse { get; set; }
        public bool DataPoint3Analyzed { get; set; }
        public string DataPoint3AnalysisResponse { get; set; }
        public bool DataPoint4Analyzed { get; set; }
        public string DataPoint4AnalysisResponse { get; set; }
    }
}

[sophialagerkranspandey]

Hi, yes there is a better way we recommend. We recommend you use the Semantic Kernel Process Framework for this scenario: https://learn.microsoft.com/en-us/semantic-kernel/frameworks/process/process-framework

[bux]

Sorry. I've just realized, I didn't mention that I am already using the process framework. The steps I mentioned above are part of a process.

[sophialagerkranspandey]

Thanks, tagging @alliscode

[bux]

I've added example code