From 5c3c2fe63460d3b7d8973341ed0eeaa8c89e3d1f Mon Sep 17 00:00:00 2001 From: Chris Gillum Date: Tue, 9 Sep 2025 08:23:00 -0700 Subject: [PATCH] .NET: Multiple fixes in Microsoft.Agents.Orchestration to make handoffs work end-to-end (#568) * Multiple fixes in Microsoft.Agents.Orchestration: - Enable multi-turn (via RunAsync) interactions for all orchestrating agents - Fix tool calling in HandoffOrchestration (multiple issues) - Fixes for chat history serialization involving tool calls * GHCP PR feedback * Fix issue with returning old responses --------- Co-authored-by: Chris <66376200+crickman@users.noreply.github.com> --- .../ConcurrentOrchestration.cs | 7 ++-- .../GroupChat/GroupChatOrchestration.cs | 8 +++-- .../Handoffs/HandoffOrchestration.cs | 33 ++++++++++--------- .../OrchestratingAgent.cs | 5 +-- .../OrchestrationJsonContext.cs | 2 ++ .../SequentialOrchestration.cs | 7 ++-- .../OrchestrationResultTests.cs | 2 +- 7 files changed, 39 insertions(+), 25 deletions(-) diff --git a/dotnet/src/Microsoft.Agents.Orchestration/ConcurrentOrchestration.cs b/dotnet/src/Microsoft.Agents.Orchestration/ConcurrentOrchestration.cs index 670731efe8..3953b7fd0f 100644 --- a/dotnet/src/Microsoft.Agents.Orchestration/ConcurrentOrchestration.cs +++ b/dotnet/src/Microsoft.Agents.Orchestration/ConcurrentOrchestration.cs @@ -61,10 +61,13 @@ public partial class ConcurrentOrchestration : OrchestratingAgent this.ResumeAsync(messages, new AgentRunResponse?[this.Agents.Count], context, cancellationToken); /// - protected override Task ResumeCoreAsync(JsonElement checkpointState, OrchestratingAgentContext context, CancellationToken cancellationToken) + protected override Task ResumeCoreAsync(JsonElement checkpointState, IReadOnlyCollection newMessages, OrchestratingAgentContext context, CancellationToken cancellationToken) { var state = checkpointState.Deserialize(OrchestrationJsonContext.Default.ConcurrentState) ?? throw new InvalidOperationException("The checkpoint state is invalid."); - return this.ResumeAsync(state.Messages, state.Completed, context, cancellationToken); + + // Append the new messages to the checkpoint state + List allMessages = [.. state.Messages, .. newMessages]; + return this.ResumeAsync(allMessages, state.Completed, context, cancellationToken); } /// diff --git a/dotnet/src/Microsoft.Agents.Orchestration/GroupChat/GroupChatOrchestration.cs b/dotnet/src/Microsoft.Agents.Orchestration/GroupChat/GroupChatOrchestration.cs index 46f258715d..0f0973e768 100644 --- a/dotnet/src/Microsoft.Agents.Orchestration/GroupChat/GroupChatOrchestration.cs +++ b/dotnet/src/Microsoft.Agents.Orchestration/GroupChat/GroupChatOrchestration.cs @@ -51,10 +51,14 @@ public sealed partial class GroupChatOrchestration : OrchestratingAgent } /// - protected override Task ResumeCoreAsync(JsonElement checkpointState, OrchestratingAgentContext context, CancellationToken cancellationToken) + protected override Task ResumeCoreAsync(JsonElement checkpointState, IReadOnlyCollection newMessages, OrchestratingAgentContext context, CancellationToken cancellationToken) { var state = checkpointState.Deserialize(OrchestrationJsonContext.Default.GroupChatState) ?? throw new InvalidOperationException("The checkpoint state is invalid."); - return this.ResumeAsync(state.AllMessages, state.OriginalMessageCount, context, cancellationToken); + + // Append the new messages to the checkpoint state + List allMessages = [.. state.AllMessages, .. newMessages]; + + return this.ResumeAsync(allMessages, allMessages.Count, context, cancellationToken); } private async Task ResumeAsync( diff --git a/dotnet/src/Microsoft.Agents.Orchestration/Handoffs/HandoffOrchestration.cs b/dotnet/src/Microsoft.Agents.Orchestration/Handoffs/HandoffOrchestration.cs index d1c9c5568e..321e71b4a2 100644 --- a/dotnet/src/Microsoft.Agents.Orchestration/Handoffs/HandoffOrchestration.cs +++ b/dotnet/src/Microsoft.Agents.Orchestration/Handoffs/HandoffOrchestration.cs @@ -52,26 +52,28 @@ public sealed partial class HandoffOrchestration : OrchestratingAgent } /// - protected override Task ResumeCoreAsync(JsonElement checkpointState, OrchestratingAgentContext context, CancellationToken cancellationToken) + protected override Task ResumeCoreAsync(JsonElement checkpointState, IReadOnlyCollection newMessages, OrchestratingAgentContext context, CancellationToken cancellationToken) { var state = checkpointState.Deserialize(OrchestrationJsonContext.Default.HandoffState) ?? throw new InvalidOperationException("The checkpoint state is invalid."); AIAgent? nextAgent = null; - foreach (var agent in this.Agents) + if (state.NextAgent is null) { - if (agent.Id == state.NextAgent) + nextAgent = this._handoffs.InitialAgent; + } + else + { + nextAgent = this.Agents.FirstOrDefault(a => a.Id == state.NextAgent); + if (nextAgent is null) { - nextAgent = agent; - break; + Throw.InvalidOperationException($"The next agent '{state.NextAgent}' is not defined in the orchestration."); } } - if (nextAgent is null) - { - Throw.InvalidOperationException($"The next agent '{state.NextAgent}' is not defined in the orchestration."); - } + // Append the new messages to the checkpoint state + List allMessages = [.. state.AllMessages, .. newMessages]; - return this.ResumeAsync(nextAgent, state.AllMessages, state.OriginalMessageCount, context, cancellationToken); + return this.ResumeAsync(nextAgent, allMessages, allMessages.Count, context, cancellationToken); } /// @@ -144,7 +146,7 @@ public sealed partial class HandoffOrchestration : OrchestratingAgent private static void RemoveHandoffFunctionCalls(AgentRunResponse response, List handoffTools) { HashSet? removeToolNames = null; - HashSet? callIds = null; + HashSet? handoffCallIds = null; foreach (var message in response.Messages) { @@ -153,23 +155,22 @@ public sealed partial class HandoffOrchestration : OrchestratingAgent if (message.Contents[i] is FunctionCallContent fcc) { removeToolNames ??= [.. handoffTools.Select(t => t.Name)]; - (callIds ??= new()).Add(fcc.CallId); - if (removeToolNames.Contains(fcc.Name)) { + (handoffCallIds ??= []).Add(fcc.CallId); message.Contents.RemoveAt(i); } } } } - if (callIds is not null) + if (handoffCallIds is not null) { foreach (var message in response.Messages) { for (int i = message.Contents.Count - 1; i >= 0; i--) { - if (message.Contents[i] is FunctionResultContent frc && callIds.Contains(frc.CallId)) + if (message.Contents[i] is FunctionResultContent frc && handoffCallIds.Contains(frc.CallId)) { message.Contents.RemoveAt(i); } @@ -215,7 +216,7 @@ public sealed partial class HandoffOrchestration : OrchestratingAgent static void Terminate() { - if (FunctionInvokingChatClient.CurrentContext is not { } ctx) + if (NewFunctionInvokingChatClient.CurrentContext is not { } ctx) { throw new NotSupportedException($"The agent is not configured with a {nameof(FunctionInvokingChatClient)}. Cease execution."); } diff --git a/dotnet/src/Microsoft.Agents.Orchestration/OrchestratingAgent.cs b/dotnet/src/Microsoft.Agents.Orchestration/OrchestratingAgent.cs index 733003d8c0..dfbe7bddfe 100644 --- a/dotnet/src/Microsoft.Agents.Orchestration/OrchestratingAgent.cs +++ b/dotnet/src/Microsoft.Agents.Orchestration/OrchestratingAgent.cs @@ -135,7 +135,7 @@ public abstract partial class OrchestratingAgent : AIAgent JsonElement? checkpoint = await this.ReadCheckpointAsync(context, cancellationToken).ConfigureAwait(false); Task completion = checkpoint is null ? this.RunCoreAsync(messages, context, cancellationToken) : - this.ResumeCoreAsync(checkpoint.Value, context, cancellationToken); + this.ResumeCoreAsync(checkpoint.Value, messages, context, cancellationToken); if (logger.IsEnabled(LogLevel.Trace)) { @@ -157,9 +157,10 @@ public abstract partial class OrchestratingAgent : AIAgent /// Resumes processing of the orchestration. /// /// The last checkpoint state available from which to resume the operation. + /// The new messages to be processed in addition to the checkpoint state. /// The context for this operation. /// A cancellation token that can be used to cancel the operation. - protected abstract Task ResumeCoreAsync(JsonElement checkpointState, OrchestratingAgentContext context, CancellationToken cancellationToken); + protected abstract Task ResumeCoreAsync(JsonElement checkpointState, IReadOnlyCollection newMessages, OrchestratingAgentContext context, CancellationToken cancellationToken); /// /// Runs the agent with input messages and respond with both streamed and regular messages. diff --git a/dotnet/src/Microsoft.Agents.Orchestration/OrchestrationJsonContext.cs b/dotnet/src/Microsoft.Agents.Orchestration/OrchestrationJsonContext.cs index 1ef15b385a..5066f3e6dc 100644 --- a/dotnet/src/Microsoft.Agents.Orchestration/OrchestrationJsonContext.cs +++ b/dotnet/src/Microsoft.Agents.Orchestration/OrchestrationJsonContext.cs @@ -1,5 +1,6 @@ // Copyright (c) Microsoft. All rights reserved. +using System.Text.Json; using System.Text.Json.Serialization; namespace Microsoft.Agents.Orchestration; @@ -8,4 +9,5 @@ namespace Microsoft.Agents.Orchestration; [JsonSerializable(typeof(ConcurrentOrchestration.ConcurrentState))] [JsonSerializable(typeof(GroupChatOrchestration.GroupChatState))] [JsonSerializable(typeof(HandoffOrchestration.HandoffState))] +[JsonSerializable(typeof(JsonElement))] internal sealed partial class OrchestrationJsonContext : JsonSerializerContext; diff --git a/dotnet/src/Microsoft.Agents.Orchestration/SequentialOrchestration.cs b/dotnet/src/Microsoft.Agents.Orchestration/SequentialOrchestration.cs index 6f79bb4129..1c113cfbde 100644 --- a/dotnet/src/Microsoft.Agents.Orchestration/SequentialOrchestration.cs +++ b/dotnet/src/Microsoft.Agents.Orchestration/SequentialOrchestration.cs @@ -32,10 +32,13 @@ public sealed partial class SequentialOrchestration : OrchestratingAgent this.ResumeAsync(0, messages, context, cancellationToken); /// - protected override Task ResumeCoreAsync(JsonElement checkpointState, OrchestratingAgentContext context, CancellationToken cancellationToken) + protected override Task ResumeCoreAsync(JsonElement checkpointState, IReadOnlyCollection newMessages, OrchestratingAgentContext context, CancellationToken cancellationToken) { var state = checkpointState.Deserialize(OrchestrationJsonContext.Default.SequentialState) ?? throw new InvalidOperationException("The checkpoint state is invalid."); - return this.ResumeAsync(state.Index, state.Messages, context, cancellationToken); + + // Append the new messages to the checkpoint state + List allMessages = [.. state.Messages, .. newMessages]; + return this.ResumeAsync(state.Index, allMessages, context, cancellationToken); } /// diff --git a/dotnet/tests/Microsoft.Agents.Orchestration.UnitTests/OrchestrationResultTests.cs b/dotnet/tests/Microsoft.Agents.Orchestration.UnitTests/OrchestrationResultTests.cs index bbc79b661c..9d9f238cca 100644 --- a/dotnet/tests/Microsoft.Agents.Orchestration.UnitTests/OrchestrationResultTests.cs +++ b/dotnet/tests/Microsoft.Agents.Orchestration.UnitTests/OrchestrationResultTests.cs @@ -83,7 +83,7 @@ public class OrchestrationResultTests protected override Task RunCoreAsync(IReadOnlyCollection messages, OrchestratingAgentContext context, CancellationToken cancellationToken) => throw new NotSupportedException(); - protected override Task ResumeCoreAsync(JsonElement checkpointState, OrchestratingAgentContext context, CancellationToken cancellationToken) => + protected override Task ResumeCoreAsync(JsonElement checkpointState, IReadOnlyCollection newMessages, OrchestratingAgentContext context, CancellationToken cancellationToken) => throw new NotSupportedException(); }