From 738866f4fe8a950c7f6321a9d11a41f418782350 Mon Sep 17 00:00:00 2001 From: Jacob Alber Date: Wed, 27 Aug 2025 18:32:27 -0400 Subject: [PATCH] Aggregate ChatMessages from AgentRunUpdates before forwarding (#515) The current implementation of AIAgentHostExecutor unwraps every incoming AgentRunResponseUpdate into a separate ChatMessage, amplifying the number of ChatMessages are actually generated, and yielding multiple messages with the same MessageId. The fix is to aggregate by MesageId, with the expectation that agents do not interleave messages with differing ids, thus every new MessageId indicates a new ChatMessage and never an old one. --- .../Specialized/AIAgentHostExecutor.cs | 34 ++++++++++++++----- 1 file changed, 25 insertions(+), 9 deletions(-) diff --git a/dotnet/src/Microsoft.Agents.Workflows/Specialized/AIAgentHostExecutor.cs b/dotnet/src/Microsoft.Agents.Workflows/Specialized/AIAgentHostExecutor.cs index 006287c023..3bde16e6e6 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Specialized/AIAgentHostExecutor.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Specialized/AIAgentHostExecutor.cs @@ -95,6 +95,8 @@ internal class AIAgentHostExecutor : Executor IAsyncEnumerable agentStream = this._agent.RunStreamingAsync(this._pendingMessages, this.EnsureThread(context)); List updates = new(); + ChatMessage? currentStreamingMessage = null; + await foreach (AgentRunResponseUpdate update in agentStream.ConfigureAwait(false)) { if (emitEvents) @@ -108,18 +110,32 @@ internal class AIAgentHostExecutor : Executor // workflow. updates.Add(update); - ChatMessage message = new(update.Role ?? ChatRole.Assistant, update.Contents) - { - AuthorName = update.AuthorName, - CreatedAt = update.CreatedAt, - MessageId = update.MessageId, - RawRepresentation = update.RawRepresentation, - AdditionalProperties = update.AdditionalProperties - }; - await context.SendMessageAsync(message).ConfigureAwait(false); + if (currentStreamingMessage == null || currentStreamingMessage.MessageId != update.MessageId) + { + await PublishCurrentMessageAsync().ConfigureAwait(false); + currentStreamingMessage = new(update.Role ?? ChatRole.Assistant, update.Contents) + { + AuthorName = update.AuthorName, + CreatedAt = update.CreatedAt, + MessageId = update.MessageId, + RawRepresentation = update.RawRepresentation, + AdditionalProperties = update.AdditionalProperties + }; + } } + await PublishCurrentMessageAsync().ConfigureAwait(false); await context.SendMessageAsync(token).ConfigureAwait(false); + + async ValueTask PublishCurrentMessageAsync() + { + if (currentStreamingMessage != null) + { + await context.SendMessageAsync(currentStreamingMessage).ConfigureAwait(false); + } + + currentStreamingMessage = null; + } } }