Skip to content

Commit 6fee23c

Browse files
crickmancrickman
authored
.Net Processes - Support Complex Type Serialization for Dapr Events and Messages (#9525)
### Motivation and Context <!-- Thank you for your contribution to the semantic-kernel repo! Please help reviewers and future users, providing the following information: 1. Why is this change required? 2. What problem does it solve? 3. What scenario does it contribute to? 4. If it fixes an open issue, please link to the issue here. --> DAPR events and messages result in runtime failure when contain anything other than the supported primitive types: https://docs.dapr.io/developing-applications/sdks/dotnet/dotnet-actors/dotnet-actors-serialization/#supported-primitive-types This means arrays, lists, and objects are not currently able to be emitted by steps. ### Description <!-- Describe your changes, the overall approach, the underlying design. These notes will help understanding how your code works. Thanks! --> Translating `ProcessMessage`, `ProcessEvent`, and `KernelProcessMessage` into JSON strings prior to data-contract serialization. As part of this, any relevant type information is included in the json payload. On deserialization, the type information is utilized to convert any `JsonElement` data values to the expected type. **Notes:** - Added unit tests (DAPR specific project) - Removed `DataContract` where not needed (and tests) - Verified DAPR demo - Attempted to minimize postfix null-supression operator as able. - Replaced `var` with explicit types for my own clarity...and left them for posterity. - Eliminated redundant namespace scoping ### Contribution Checklist <!-- Before submitting this PR, please make sure: --> - [X] The code builds clean without any errors or warnings - [X] The PR follows the [SK Contribution Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md) and the [pre-submission formatting script](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md#development-scripts) raises no violations - [X] All unit tests pass, and I have added new tests where possible - [X] I didn't break anyone 😄
1 parent 9eae969 commit 6fee23c

31 files changed

+809
-297
lines changed

‎dotnet/SK-dotnet.sln‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -349,7 +349,6 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "FunctionCalling", "Function
349349
ProjectSection(SolutionItems) = preProject
350350
src\InternalUtilities\connectors\AI\FunctionCalling\FunctionCallingUtilities.props = src\InternalUtilities\connectors\AI\FunctionCalling\FunctionCallingUtilities.props
351351
src\InternalUtilities\connectors\AI\FunctionCalling\FunctionCallsProcessor.cs = src\InternalUtilities\connectors\AI\FunctionCalling\FunctionCallsProcessor.cs
352-
src\InternalUtilities\connectors\AI\FunctionCalling\FunctionCallsProcessorLoggerExtensions.cs = src\InternalUtilities\connectors\AI\FunctionCalling\FunctionCallsProcessorLoggerExtensions.cs
353352
EndProjectSection
354353
EndProject
355354
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Connectors.Weaviate.UnitTests", "src\Connectors\Connectors.Weaviate.UnitTests\Connectors.Weaviate.UnitTests.csproj", "{E8FC97B0-B417-4A90-993C-B8AA9223B058}"
@@ -411,6 +410,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Process.Utilities.UnitTests
411410
EndProject
412411
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "GettingStartedWithVectorStores", "samples\GettingStartedWithVectorStores\GettingStartedWithVectorStores.csproj", "{8C3DE41C-E2C8-42B9-8638-574F8946EB0E}"
413412
EndProject
413+
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Process.Runtime.Dapr.UnitTests", "src\Experimental\Process.Runtime.Dapr.UnitTests\Process.Runtime.Dapr.UnitTests.csproj", "{DB58FDD0-308E-472F-BFF5-508BC64C727E}"
414+
EndProject
414415
Global
415416
GlobalSection(SolutionConfigurationPlatforms) = preSolution
416417
Debug|Any CPU = Debug|Any CPU
@@ -1078,6 +1079,12 @@ Global
10781079
{8C3DE41C-E2C8-42B9-8638-574F8946EB0E}.Publish|Any CPU.Build.0 = Debug|Any CPU
10791080
{8C3DE41C-E2C8-42B9-8638-574F8946EB0E}.Release|Any CPU.ActiveCfg = Release|Any CPU
10801081
{8C3DE41C-E2C8-42B9-8638-574F8946EB0E}.Release|Any CPU.Build.0 = Release|Any CPU
1082+
{DB58FDD0-308E-472F-BFF5-508BC64C727E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
1083+
{DB58FDD0-308E-472F-BFF5-508BC64C727E}.Debug|Any CPU.Build.0 = Debug|Any CPU
1084+
{DB58FDD0-308E-472F-BFF5-508BC64C727E}.Publish|Any CPU.ActiveCfg = Debug|Any CPU
1085+
{DB58FDD0-308E-472F-BFF5-508BC64C727E}.Publish|Any CPU.Build.0 = Debug|Any CPU
1086+
{DB58FDD0-308E-472F-BFF5-508BC64C727E}.Release|Any CPU.ActiveCfg = Release|Any CPU
1087+
{DB58FDD0-308E-472F-BFF5-508BC64C727E}.Release|Any CPU.Build.0 = Release|Any CPU
10811088
EndGlobalSection
10821089
GlobalSection(SolutionProperties) = preSolution
10831090
HideSolutionNode = FALSE
@@ -1226,6 +1233,7 @@ Global
12261233
{39EAB599-742F-417D-AF80-95F90376BB18} = {831DDCA2-7D2C-4C31-80DB-6BDB3E1F7AE0}
12271234
{DAC54048-A39A-4739-8307-EA5A291F2EA0} = {0D8C6358-5DAA-4EA6-A924-C268A9A21BC9}
12281235
{8C3DE41C-E2C8-42B9-8638-574F8946EB0E} = {FA3720F1-C99A-49B2-9577-A940257098BF}
1236+
{DB58FDD0-308E-472F-BFF5-508BC64C727E} = {0D8C6358-5DAA-4EA6-A924-C268A9A21BC9}
12291237
EndGlobalSection
12301238
GlobalSection(ExtensibilityGlobals) = postSolution
12311239
SolutionGuid = {FBDC56A3-86AD-4323-AA0F-201E59123B83}

‎dotnet/src/Experimental/Process.Abstractions/KernelProcessError.cs‎

Lines changed: 20 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
// Copyright (c) Microsoft. All rights reserved.
22
using System;
3-
using System.Runtime.Serialization;
43

54
namespace Microsoft.SemanticKernel;
65

@@ -10,30 +9,37 @@ namespace Microsoft.SemanticKernel;
109
/// <remarks>
1110
/// Initializes a new instance of the <see cref="KernelProcessError"/> class.
1211
/// </remarks>
13-
/// <param name="Type">The exception type name</param>
14-
/// <param name="Message">The exception message (<see cref="Exception.Message"/></param>
15-
/// <param name="StackTrace">The exception stack-trace (<see cref="Exception.StackTrace"/></param>
16-
[DataContract]
17-
public sealed record KernelProcessError(
18-
[property:DataMember]
19-
string Type,
20-
[property:DataMember]
21-
string Message,
22-
[property:DataMember]
23-
string? StackTrace)
12+
public sealed record KernelProcessError
2413
{
14+
/// <summary>
15+
///The exception type name.
16+
/// </summary>
17+
public string Type { get; init; } = string.Empty;
18+
19+
/// <summary>
20+
/// The exception message (<see cref="Exception.Message"/>.
21+
/// </summary>
22+
public string Message { get; init; } = string.Empty;
23+
24+
/// <summary>
25+
/// The exception stack-trace (<see cref="Exception.StackTrace"/>.
26+
/// </summary>
27+
public string? StackTrace { get; init; }
28+
2529
/// <summary>
2630
/// The inner failure, when exists, as <see cref="KernelProcessError"/>.
2731
/// </summary>
28-
[DataMember]
2932
public KernelProcessError? InnerError { get; init; }
3033

3134
/// <summary>
3235
/// Factory method to create a <see cref="KernelProcessError"/> from a source <see cref="Exception"/> object.
3336
/// </summary>
3437
public static KernelProcessError FromException(Exception ex) =>
35-
new(ex.GetType().Name, ex.Message, ex.StackTrace)
38+
new()
3639
{
40+
Type = ex.GetType().Name,
41+
Message = ex.Message,
42+
StackTrace = ex.StackTrace,
3743
InnerError = ex.InnerException is not null ? FromException(ex.InnerException) : null
3844
};
3945
}

‎dotnet/src/Experimental/Process.Abstractions/KernelProcessEvent.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,19 +3,19 @@
33
namespace Microsoft.SemanticKernel;
44

55
/// <summary>
6-
/// An class representing an event that can be emitted from a <see cref="KernelProcessStep"/>. This type is convertible to and from CloudEvents.
6+
/// A class representing an event that can be emitted from a <see cref="KernelProcessStep"/>. This type is convertible to and from CloudEvents.
77
/// </summary>
88
public sealed record KernelProcessEvent
99
{
1010
/// <summary>
1111
/// The unique identifier for the event.
1212
/// </summary>
13-
public string? Id { get; set; }
13+
public string Id { get; init; } = string.Empty;
1414

1515
/// <summary>
1616
/// An optional data payload associated with the event.
1717
/// </summary>
18-
public object? Data { get; set; }
18+
public object? Data { get; init; }
1919

2020
/// <summary>
2121
/// The visibility of the event. Defaults to <see cref="KernelProcessEventVisibility.Internal"/>.

‎dotnet/src/Experimental/Process.LocalRuntime/LocalProcess.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -149,7 +149,7 @@ internal override async Task HandleMessageAsync(ProcessMessage message)
149149
{
150150
// Create the external event that will be used to start the nested process. Since this event came
151151
// from outside this processes, we set the visibility to internal so that it's not emitted back out again.
152-
var nestedEvent = new KernelProcessEvent() { Id = eventId, Data = message.TargetEventData, Visibility = KernelProcessEventVisibility.Internal };
152+
KernelProcessEvent nestedEvent = new() { Id = eventId, Data = message.TargetEventData, Visibility = KernelProcessEventVisibility.Internal };
153153

154154
// Run the nested process completely within a single superstep.
155155
await this.RunOnceAsync(nestedEvent, this._kernel).ConfigureAwait(false);
@@ -300,7 +300,7 @@ private void EnqueueExternalMessages(Queue<ProcessMessage> messageChannel)
300300
{
301301
while (this._externalEventChannel.Reader.TryRead(out var externalEvent))
302302
{
303-
if (this._outputEdges!.TryGetValue(externalEvent.Id!, out List<KernelProcessEdge>? edges) && edges is not null)
303+
if (this._outputEdges.TryGetValue(externalEvent.Id, out List<KernelProcessEdge>? edges) && edges is not null)
304304
{
305305
foreach (var edge in edges)
306306
{
@@ -330,7 +330,7 @@ private void EnqueueStepMessages(LocalStep step, Queue<ProcessMessage> messageCh
330330

331331
// Get the edges for the event and queue up the messages to be sent to the next steps.
332332
bool foundEdge = false;
333-
foreach (KernelProcessEdge edge in step.GetEdgeForEvent(stepEvent.Id))
333+
foreach (KernelProcessEdge edge in step.GetEdgeForEvent(stepEvent.QualifiedId))
334334
{
335335
ProcessMessage message = ProcessMessageFactory.CreateFromEdge(edge, stepEvent.Data);
336336
messageChannel.Enqueue(message);

‎dotnet/src/Experimental/Process.LocalRuntime/LocalStep.cs‎

Lines changed: 20 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -112,17 +112,9 @@ internal IEnumerable<KernelProcessEdge> GetEdgeForEvent(string eventId)
112112
/// </summary>
113113
/// <param name="processEvent">The event to emit.</param>
114114
/// <returns>A <see cref="ValueTask"/></returns>
115-
public ValueTask EmitEventAsync(KernelProcessEvent processEvent) => this.EmitEventAsync(processEvent, isError: false);
116-
117-
/// <summary>
118-
/// Emits an event from the step.
119-
/// </summary>
120-
/// <param name="processEvent">The event to emit.</param>
121-
/// <param name="isError">Flag indicating if the event being emitted is in response to a step failure</param>
122-
/// <returns>A <see cref="ValueTask"/></returns>
123-
internal ValueTask EmitEventAsync(KernelProcessEvent processEvent, bool isError)
115+
public ValueTask EmitEventAsync(KernelProcessEvent processEvent)
124116
{
125-
this.EmitEvent(new ProcessEvent(this._eventNamespace, processEvent, isError));
117+
this.EmitEvent(ProcessEvent.Create(processEvent, this._eventNamespace));
126118
return default;
127119
}
128120

@@ -193,23 +185,25 @@ internal virtual async Task HandleMessageAsync(ProcessMessage message)
193185
try
194186
{
195187
FunctionResult invokeResult = await this.InvokeFunction(function, this._kernel, arguments).ConfigureAwait(false);
196-
await this.EmitEventAsync(
197-
new KernelProcessEvent
188+
this.EmitEvent(
189+
new ProcessEvent
198190
{
199-
Id = $"{targetFunction}.OnResult",
200-
Data = invokeResult.GetValue<object>(),
201-
}).ConfigureAwait(false);
191+
Namespace = this._eventNamespace,
192+
SourceId = $"{targetFunction}.OnResult",
193+
Data = invokeResult.GetValue<object>()
194+
});
202195
}
203196
catch (Exception ex)
204197
{
205-
this._logger.LogError("Error in Step {StepName}: {ErrorMessage}", this.Name, ex.Message);
206-
await this.EmitEventAsync(
207-
new KernelProcessEvent
198+
this._logger.LogError(ex, "Error in Step {StepName}: {ErrorMessage}", this.Name, ex.Message);
199+
this.EmitEvent(
200+
new ProcessEvent
208201
{
209-
Id = $"{targetFunction}.OnError",
202+
Namespace = this._eventNamespace,
203+
SourceId = $"{targetFunction}.OnError",
210204
Data = KernelProcessError.FromException(ex),
211-
},
212-
isError: true).ConfigureAwait(false);
205+
IsError = true
206+
});
213207
}
214208
finally
215209
{
@@ -250,23 +244,18 @@ protected virtual async ValueTask InitializeStepAsync()
250244
throw new KernelException("The state object for the KernelProcessStep could not be created.").Log(this._logger);
251245
}
252246

253-
MethodInfo? methodInfo = this._stepInfo.InnerStepType.GetMethod(nameof(KernelProcessStep.ActivateAsync), [stateType]);
254-
255-
if (methodInfo is null)
256-
{
247+
MethodInfo methodInfo =
248+
this._stepInfo.InnerStepType.GetMethod(nameof(KernelProcessStep.ActivateAsync), [stateType]) ??
257249
throw new KernelException("The ActivateAsync method for the KernelProcessStep could not be found.").Log(this._logger);
258-
}
259250

260251
this._stepState = stateObject;
261252

262-
ValueTask? activateTask = (ValueTask?)methodInfo.Invoke(stepInstance, [stateObject]);
263-
if (activateTask == null)
264-
{
253+
ValueTask activateTask =
254+
(ValueTask?)methodInfo.Invoke(stepInstance, [stateObject]) ??
265255
throw new KernelException("The ActivateAsync method failed to complete.").Log(this._logger);
266-
}
267256

268257
await stepInstance.ActivateAsync(stateObject).ConfigureAwait(false);
269-
await activateTask.Value.ConfigureAwait(false);
258+
await activateTask.ConfigureAwait(false);
270259
}
271260

272261
/// <summary>
@@ -315,15 +304,4 @@ protected ProcessEvent ScopedEvent(ProcessEvent localEvent)
315304
Verify.NotNull(localEvent, nameof(localEvent));
316305
return localEvent with { Namespace = $"{this.Name}_{this.Id}" };
317306
}
318-
319-
/// <summary>
320-
/// Generates a scoped event for the step.
321-
/// </summary>
322-
/// <param name="processEvent">The event.</param>
323-
/// <returns>A <see cref="ProcessEvent"/> with the correctly scoped namespace.</returns>
324-
protected ProcessEvent ScopedEvent(KernelProcessEvent processEvent)
325-
{
326-
Verify.NotNull(processEvent, nameof(processEvent));
327-
return new ProcessEvent($"{this.Name}_{this.Id}", processEvent);
328-
}
329307
}

‎dotnet/src/Experimental/Process.Runtime.Dapr.UnitTests/KernelProcessEventSerializationTests.cs‎

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,114 @@
1+
// Copyright (c) Microsoft. All rights reserved.
2+
using System;
3+
using System.Collections.Generic;
4+
using System.IO;
5+
using Microsoft.SemanticKernel;
6+
using Microsoft.SemanticKernel.Process.Runtime;
7+
using Microsoft.SemanticKernel.Process.Serialization;
8+
using Xunit;
9+
10+
namespace SemanticKernel.Process.Dapr.Runtime.UnitTests;
11+
12+
/// <summary>
13+
/// Unit tests for the <see cref="ProcessEvent"/> class.
14+
/// </summary>
15+
public class KernelProcessEventSerializationTests
16+
{
17+
/// <summary>
18+
/// Validates that a <see cref="KernelProcessEvent"/> can be serialized and deserialized correctly
19+
/// with out an explicit type definition for <see cref="KernelProcessEvent.Data"/>
20+
/// </summary>
21+
[Fact]
22+
public void VerifySerializeEventSingleTest()
23+
{
24+
// Arrange, Act & Assert
25+
VerifyContainerSerialization([new() { Id = "Test", Data = 3 }]);
26+
VerifyContainerSerialization([new() { Id = "Test", Data = "test" }]);
27+
VerifyContainerSerialization([new() { Id = "Test", Data = Guid.NewGuid() }]);
28+
VerifyContainerSerialization([new() { Id = "Test", Data = new int[] { 1, 2, 3, 4 } }]);
29+
VerifyContainerSerialization([new() { Id = "Test", Data = new ComplexData { Id = "test", Value = 3 } }]);
30+
VerifyContainerSerialization([new() { Id = "testid", Data = KernelProcessError.FromException(new InvalidOperationException()) }]);
31+
}
32+
33+
/// <summary>
34+
/// Validates that a list <see cref="KernelProcessEvent"/> can be serialized and deserialized correctly
35+
/// with out varying types assigned to for <see cref="KernelProcessEvent.Data"/>
36+
/// </summary>
37+
[Fact]
38+
public void VerifySerializeEventMixedTest()
39+
{
40+
// Arrange, Act & Assert
41+
VerifyContainerSerialization(
42+
[
43+
new() { Id = "Test", Data = 3 },
44+
new() { Id = "Test", Data = "test" },
45+
new() { Id = "Test", Data = Guid.NewGuid() },
46+
new() { Id = "Test", Data = new int[] { 1, 2, 3, 4 } },
47+
new() { Id = "Test", Data = new ComplexData { Id = "test", Value = 3 } },
48+
new() { Id = "testid", Data = KernelProcessError.FromException(new InvalidOperationException()) },
49+
]);
50+
}
51+
52+
/// <summary>
53+
/// Validates that a list <see cref="KernelProcessEvent"/> can be serialized and deserialized correctly
54+
/// with out varying types assigned to for <see cref="KernelProcessEvent.Data"/>
55+
/// </summary>
56+
[Fact]
57+
public void VerifyDataContractSerializationTest()
58+
{
59+
// Arrange
60+
KernelProcessEvent[] processEvents =
61+
[
62+
new() { Id = "Test", Data = 3 },
63+
new() { Id = "Test", Data = "test" },
64+
new() { Id = "Test", Data = Guid.NewGuid() },
65+
new() { Id = "Test", Data = new int[] { 1, 2, 3, 4 } },
66+
new() { Id = "Test", Data = new ComplexData { Id = "test", Value = 3 } },
67+
new() { Id = "testid", Data = KernelProcessError.FromException(new InvalidOperationException()) },
68+
];
69+
List<string> jsonEvents = [];
70+
foreach (KernelProcessEvent processEvent in processEvents)
71+
{
72+
jsonEvents.Add(KernelProcessEventSerializer.ToJson(processEvent));
73+
}
74+
75+
// Act
76+
using MemoryStream stream = new();
77+
jsonEvents.Serialize(stream);
78+
stream.Position = 0;
79+
80+
List<string>? copy = stream.Deserialize<List<string>>();
81+
82+
// Assert
83+
Assert.NotNull(copy);
84+
85+
// Act
86+
IList<KernelProcessEvent> copiedEvents = KernelProcessEventSerializer.ToKernelProcessEvents(jsonEvents);
87+
88+
// Assert
89+
Assert.Equivalent(processEvents, copiedEvents);
90+
}
91+
92+
private static void VerifyContainerSerialization(KernelProcessEvent[] processEvents)
93+
{
94+
// Arrange
95+
List<string> jsonEvents = [];
96+
foreach (KernelProcessEvent processEvent in processEvents)
97+
{
98+
jsonEvents.Add(KernelProcessEventSerializer.ToJson(processEvent));
99+
}
100+
101+
// Act
102+
IList<KernelProcessEvent> copiedEvents = KernelProcessEventSerializer.ToKernelProcessEvents(jsonEvents);
103+
104+
// Assert
105+
Assert.Equivalent(processEvents, copiedEvents);
106+
}
107+
108+
internal sealed class ComplexData
109+
{
110+
public string Id { get; init; } = string.Empty;
111+
112+
public int Value { get; init; }
113+
}
114+
}

0 commit comments

Comments
 (0)