.Net Processes - Map Step Feature

dotnet/samples/GettingStartedWithProcesses/GettingStartedWithProcesses.csproj

@@ -56,7 +56,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <EmbeddedResource Include="Resources\*">
+    <EmbeddedResource Include="..\LearnResources\Resources\Grimms-*">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </EmbeddedResource>
   </ItemGroup>

dotnet/samples/GettingStartedWithProcesses/Step05/Step05_MapReduce.cs

@@ -0,0 +1,227 @@
+// Copyright (c) Microsoft. All rights reserved.
+using System.Text;
+using Microsoft.SemanticKernel;
+
+namespace Step05;
+
+/// <summary>
+/// DEV HARNESS
+/// </summary>
+public class Step05_MapReduce : BaseTest
+{
+    // Target Open AI Services
+    protected override bool ForceOpenAI => true;
+
+    /// <summary>
+    /// Factor to increase the scale of the content processed.
+    /// </summary>
+    private const int ScaleFactor = 100;
+
+    private readonly string _sourceContent;
+
+    public Step05_MapReduce(ITestOutputHelper output)
+         : base(output, redirectSystemConsoleOutput: true)
+    {
+        StringBuilder content = new();
+
+        for (int count = 0; count < ScaleFactor; ++count)
+        {
+            content.AppendLine(File.ReadAllText("Grimms-The-King-of-the-Golden-Mountain.txt"));
+            content.AppendLine(File.ReadAllText("Grimms-The-Water-of-Life.txt"));
+            content.AppendLine(File.ReadAllText("Grimms-The-White-Snake.txt"));
+        }
+
+        this._sourceContent = content.ToString().ToUpperInvariant();
+    }
+
+    [Fact]
+    public async Task RunMapReduceAsync()
+    {
+        KernelProcess process = SetupMapReduceProcess(nameof(RunMapReduceAsync), "Start");
+        Kernel kernel = new();
+        using LocalKernelProcessContext localProcess =
+            await process.StartAsync(
+                kernel,
+                new KernelProcessEvent
+                {
+                    Id = "Start",
+                    Data = this._sourceContent,
+                });
+
+        Dictionary<string, int> results = (Dictionary<string, int>?)kernel.Data[ResultStep.ResultKey] ?? [];
+        foreach (var result in results)
+        {
+            Console.WriteLine($"{result.Key}: {result.Value}");
+        }
+    }
+
+    private KernelProcess SetupMapReduceProcess(string processName, string inputEventId)
+    {
+        ProcessBuilder process = new(processName);
+
+        ProcessStepBuilder chunkStep = process.AddStepFromType<ChunkStep>();
+        process
+            .OnInputEvent(inputEventId)
+            .SendEventTo(new ProcessFunctionTargetBuilder(chunkStep));
+
+        ProcessStepBuilder countStep = process.AddStepFromType<CountStep>();
+        ProcessMapBuilder mapStep = process.AddMapForTarget(new ProcessFunctionTargetBuilder(countStep));
+        chunkStep
+            .OnEvent(ChunkStep.EventId)
+            .SendEventTo(mapStep);
+
+        ProcessStepBuilder resultStep = process.AddStepFromType<ResultStep>();
+        mapStep
+            .OnEvent(CountStep.EventId)
+            .SendEventTo(new ProcessFunctionTargetBuilder(resultStep));
+
+        return process.Build();
+    }
+
+    private sealed class ChunkStep : KernelProcessStep
+    {
+        public const string EventId = "ChunkComplete";
+
+        [KernelFunction]
+        public async ValueTask ChunkAsync(KernelProcessStepContext context, string content)
+        {
+            int chunkSize = content.Length / Environment.ProcessorCount;
+            string[] chunks = ChunkContent(content, chunkSize).ToArray();
+
+            await context.EmitEventAsync(new() { Id = EventId, Data = chunks });
+        }
+
+        private IEnumerable<string> ChunkContent(string content, int chunkSize)
+        {
+            for (int index = 0; index < content.Length; index += chunkSize)
+            {
+                yield return content.Substring(index, Math.Min(chunkSize, content.Length - index));
+            }
+        }
+    }
+
+    private sealed class CountStep : KernelProcessStep
+    {
+        public const string EventId = "CountComplete";
+
+        [KernelFunction]
+        public async ValueTask ComputeAsync(KernelProcessStepContext context, string chunk)
+        {
+            Dictionary<string, int> counts = [];
+
+            string[] words = chunk.Split([' ', '\n', '\r', '.', ',', '’'], StringSplitOptions.RemoveEmptyEntries);
+            foreach (string word in words)
+            {
+                if (s_notInteresting.Contains(word))
+                {
+                    continue;
+                }
+
+                counts.TryGetValue(word.Trim(), out int count);
+                counts[word] = ++count;
+            }
+
+            await context.EmitEventAsync(new() { Id = EventId, Data = counts });
+        }
+    }
+
+    private sealed class ResultStep : KernelProcessStep
+    {
+        public const string ResultKey = "WordCount";
+
+        [KernelFunction]
+        public async ValueTask ComputeAsync(KernelProcessStepContext context, IList<Dictionary<string, int>> results, Kernel kernel)
+        {
+            Dictionary<string, int> totals = [];
+
+            foreach (Dictionary<string, int> result in results)
+            {
+                foreach (KeyValuePair<string, int> pair in result)
+                {
+                    totals.TryGetValue(pair.Key, out int count);
+                    totals[pair.Key] = count + pair.Value;
+                }
+            }
+
+            var sorted =
+                from kvp in totals
+                orderby kvp.Value descending
+                select kvp;
+
+            kernel.Data[ResultKey] = sorted.Take(10).ToDictionary(kvp => kvp.Key, kvp => kvp.Value);
+        }
+    }
+
+    private static readonly HashSet<string> s_notInteresting =
+        [
+            "A",
+            "ALL",
+            "AN",
+            "AND",
+            "AS",
+            "AT",
+            "BE",
+            "BEFORE",
+            "BUT",
+            "BY",
+            "CAME",
+            "COULD",
+            "FOR",
+            "GO",
+            "HAD",
+            "HAVE",
+            "HE",
+            "HER",
+            "HIM",
+            "HIMSELF",
+            "HIS",
+            "HOW",
+            "I",
+            "IF",
+            "IN",
+            "INTO",
+            "IS",
+            "IT",
+            "ME",
+            "MUST",
+            "MY",
+            "NO",
+            "NOT",
+            "NOW",
+            "OF",
+            "ON",
+            "ONCE",
+            "ONE",
+            "ONLY",
+            "OUT",
+            "S",
+            "SAID",
+            "SAW",
+            "SET",
+            "SHE",
+            "SHOULD",
+            "SO",
+            "THAT",
+            "THE",
+            "THEM",
+            "THEN",
+            "THEIR",
+            "THERE",
+            "THEY",
+            "THIS",
+            "TO",
+            "VERY",
+            "WAS",
+            "WENT",
+            "WERE",
+            "WHAT",
+            "WHEN",
+            "WHO",
+            "WILL",
+            "WITH",
+            "WOULD",
+            "UP",
+            "UPON",
+            "YOU",
+        ];
+}

dotnet/samples/GettingStartedWithProcesses/Utilities/ProcessStateMetadataUtilities.cs

@@ -62,7 +62,7 @@ private static void StoreProcessStateLocally(KernelProcessStateMetadata processS
             throw new KernelException($"Filepath for process {processStateInfo.Name} does not have .json extension");
         }
 
-        var content = JsonSerializer.Serialize<KernelProcessStepStateMetadata>(processStateInfo, s_jsonOptions);
+        string content = JsonSerializer.Serialize(processStateInfo, s_jsonOptions);
         Console.WriteLine($"Process State: \n{content}");
         Console.WriteLine($"Saving Process State Locally: \n{Path.GetFullPath(fullFilepath)}");
         File.WriteAllText(fullFilepath, content);

dotnet/src/Experimental/Process.Abstractions/KernelProcessMap.cs

@@ -0,0 +1,31 @@
+// Copyright (c) Microsoft. All rights reserved.
+using System.Collections.Generic;
+
+namespace Microsoft.SemanticKernel;
+
+/// <summary>
+/// A serializable representation of a ProcessMap.
+/// </summary>
+public sealed record KernelProcessMap : KernelProcessStepInfo
+{
+    /// <summary>
+    /// The map operation.
+    /// </summary>
+    public KernelProcess Operation { get; }
+
+    /// <summary>
+    /// Creates a new instance of the <see cref="KernelProcess"/> class.
+    /// </summary>
+    /// <param name="state">The process state.</param>
+    /// <param name="operation">The map operation.</param>
+    /// <param name="edges">The edges for the map.</param>
+    public KernelProcessMap(KernelProcessMapState state, KernelProcess operation, Dictionary<string, List<KernelProcessEdge>> edges)
+        : base(typeof(KernelProcessMap), state, edges)
+    {
+        Verify.NotNull(operation, nameof(operation));
+        Verify.NotNullOrWhiteSpace(state.Name, $"{nameof(state)}.{nameof(KernelProcessMapState.Name)}");
+        Verify.NotNullOrWhiteSpace(state.Id, $"{nameof(state)}.{nameof(KernelProcessMapState.Id)}");
+
+        this.Operation = operation;
+    }
+}

dotnet/src/Experimental/Process.Abstractions/KernelProcessMapState.cs

@@ -0,0 +1,24 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Runtime.Serialization;
+
+namespace Microsoft.SemanticKernel;
+
+/// <summary>
+/// Represents the state of a <see cref="KernelProcessMap"/>.
+/// </summary>
+[DataContract]
+public sealed record KernelProcessMapState : KernelProcessStepState
+{
+    /// <summary>
+    /// Initializes a new instance of the <see cref="KernelProcessMapState"/> class.
+    /// </summary>
+    /// <param name="name">The name of the associated <see cref="KernelProcessMap"/></param>
+    /// <param name="version">version id of the process step state</param>
+    /// <param name="id">The Id of the associated <see cref="KernelProcessMap"/></param>
+    public KernelProcessMapState(string name, string version, string id)
+        : base(name, version, id)
+    {
+        Verify.NotNullOrWhiteSpace(id, nameof(id));
+    }
+}

dotnet/src/Experimental/Process.Abstractions/KernelProcessState.cs

@@ -5,7 +5,7 @@
 namespace Microsoft.SemanticKernel;
 
 /// <summary>
-/// Represents the state of a process.
+/// Represents the state of a <see cref="KernelProcess"/>.
 /// </summary>
 [DataContract]
 public sealed record KernelProcessState : KernelProcessStepState

dotnet/src/Experimental/Process.Abstractions/KernelProcessStep.cs

@@ -25,6 +25,6 @@ public virtual ValueTask ActivateAsync(KernelProcessStepState state)
     /// <inheritdoc/>
     public virtual ValueTask ActivateAsync(KernelProcessStepState<TState> state)
     {
-        return default;
+        return base.ActivateAsync(state);
     }
 }

dotnet/src/Experimental/Process.Abstractions/KernelProcessStepContext.cs

@@ -42,6 +42,8 @@ public ValueTask EmitEventAsync(
         object? data = null,
         KernelProcessEventVisibility visibility = KernelProcessEventVisibility.Internal)
     {
+        Verify.NotNullOrWhiteSpace(eventId, nameof(eventId));
+
         return this._stepMessageChannel.EmitEventAsync(
             new KernelProcessEvent
             {

dotnet/src/Experimental/Process.Abstractions/KernelProcessStepState.cs

@@ -62,8 +62,8 @@ internal static void RegisterDerivedType(Type derivedType)
     /// <param name="id">The Id of the associated <see cref="KernelProcessStep"/></param>
     public KernelProcessStepState(string name, string version, string? id = null)
     {
-        Verify.NotNullOrWhiteSpace(name);
-        Verify.NotNullOrWhiteSpace(version);
+        Verify.NotNullOrWhiteSpace(name, nameof(name));
+        Verify.NotNullOrWhiteSpace(version, nameof(version));
 
         this.Id = id;
         this.Name = name;

dotnet/src/Experimental/Process.Abstractions/Models/KernelProcessMapStateMetadata.cs

@@ -0,0 +1,18 @@
+// Copyright (c) Microsoft. All rights reserved.
+using System.Runtime.Serialization;
+using System.Text.Json.Serialization;
+
+namespace Microsoft.SemanticKernel.Process.Models;
+
+/// <summary>
+/// Process state used for State Persistence serialization
+/// </summary>
+public sealed record class KernelProcessMapStateMetadata : KernelProcessStepStateMetadata
+{
+    /// <summary>
+    /// Process State of Steps if provided
+    /// </summary>
+    [DataMember]
+    [JsonPropertyName("mapState")]
+    public KernelProcessStateMetadata? OperationState { get; set; }
+}

dotnet/src/Experimental/Process.Abstractions/Models/KernelProcessStepStateMetadata.cs

@@ -11,6 +11,7 @@ namespace Microsoft.SemanticKernel.Process.Models;
 /// </summary>
 [JsonPolymorphic(TypeDiscriminatorPropertyName = "$type", UnknownDerivedTypeHandling = JsonUnknownDerivedTypeHandling.FallBackToNearestAncestor)]
 [JsonDerivedType(typeof(KernelProcessStepStateMetadata), typeDiscriminator: nameof(ProcessConstants.SupportedComponents.Step))]
+[JsonDerivedType(typeof(KernelProcessMapStateMetadata), typeDiscriminator: nameof(ProcessConstants.SupportedComponents.Map))]
 [JsonDerivedType(typeof(KernelProcessStateMetadata), typeDiscriminator: nameof(ProcessConstants.SupportedComponents.Process))]
 public record class KernelProcessStepStateMetadata
 {