mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
.NET: feat: Update GroupChatManager semantics to match other Orchestration patterns (#6140)
* Refactor group chat workflow to prevent message echoing and enhance checkpointing - Updated GroupChatWorkflowBuilder to disable forwarding incoming messages to prevent duplicates. - Enhanced RoundRobinGroupChatManager with checkpointing support to preserve state across executions. - Modified GroupChatHost to maintain a history of messages and track the current speaker for message broadcasting. - Implemented broadcasting logic to ensure participants receive messages from others while excluding their own responses. - Added comprehensive unit tests for group chat orchestration, including scenarios for tool approval and function calls. - Introduced a new ApprovalHarness for testing tool invocation and approval workflows. * fixup: format * Add JSON serialization support for GroupChatManagerState and RoundRobinGroupChatManagerState --------- Co-authored-by: Jacob Alber <jalber@lokitoth.com>
This commit is contained in:
Jacob Alber
co-authored by
Jacob Alber
Jacob Alber
parent
b1e9efee7e
commit
d2f79930d5
@@ -1,6 +1,8 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Microsoft.Extensions.AI;
|
||||
@@ -13,6 +15,16 @@ namespace Microsoft.Agents.AI.Workflows;
|
||||
/// </summary>
|
||||
public abstract class GroupChatManager
|
||||
{
|
||||
// The state key under which GroupChatManager persists its own (non-subclass) state on the
|
||||
// raw IWorkflowContext supplied by the hosting GroupChatHost executor.
|
||||
internal const string BaseStateKey = "GroupChatManager";
|
||||
|
||||
// Prefix automatically applied to every key a subclass writes through the wrapped context
|
||||
// supplied to OnCheckpointingAsync / OnCheckpointRestoredAsync. Keeps subclass-defined
|
||||
// state in its own namespace so it cannot collide with the host's state keys nor with
|
||||
// BaseStateKey itself.
|
||||
internal const string SubclassStateKeyPrefix = "GroupChatManager_";
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="GroupChatManager"/> class.
|
||||
/// </summary>
|
||||
@@ -48,12 +60,22 @@ public abstract class GroupChatManager
|
||||
CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>
|
||||
/// Filters the chat history before it's passed to the next agent.
|
||||
/// Filters the messages broadcast to participants for the current turn.
|
||||
/// </summary>
|
||||
/// <param name="history">The chat history to filter.</param>
|
||||
/// <remarks>
|
||||
/// Under the broadcast model, each participant maintains its own per-agent session (history)
|
||||
/// through its <see cref="Specialized.AIAgentHostExecutor"/>. The host distributes new messages
|
||||
/// (initial user input on the first turn, the most recent speaker's response on subsequent turns)
|
||||
/// to every participant — except the speaker that produced them — so every participant's session
|
||||
/// stays synchronized. This method lets the manager shape that broadcast payload (for example,
|
||||
/// to omit certain messages or to inject orchestrator-visible annotations). The full canonical
|
||||
/// conversation is still available to <see cref="SelectNextAgentAsync"/> and
|
||||
/// <see cref="ShouldTerminateAsync"/>.
|
||||
/// </remarks>
|
||||
/// <param name="history">The new messages about to be broadcast to participants this turn.</param>
|
||||
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests.
|
||||
/// The default is <see cref="CancellationToken.None"/>.</param>
|
||||
/// <returns>The filtered chat history.</returns>
|
||||
/// <returns>The filtered message list to broadcast.</returns>
|
||||
protected internal virtual ValueTask<IEnumerable<ChatMessage>> UpdateHistoryAsync(
|
||||
IReadOnlyList<ChatMessage> history,
|
||||
CancellationToken cancellationToken = default) =>
|
||||
@@ -78,4 +100,125 @@ public abstract class GroupChatManager
|
||||
{
|
||||
this.IterationCount = 0;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Invoked when the hosting group chat workflow is checkpointing, giving subclasses a chance to
|
||||
/// persist any additional state they maintain (e.g., a round-robin cursor or an LLM session).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// The default implementation is a no-op. Base-class state (currently
|
||||
/// <see cref="IterationCount"/>) is persisted automatically by the hosting
|
||||
/// <see cref="Specialized.GroupChatHost"/> before this method is invoked; subclasses do not
|
||||
/// need to call <c>base.OnCheckpointingAsync</c>.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The supplied <paramref name="context"/> is a wrapper that transparently prefixes every
|
||||
/// state key with <c>"GroupChatManager_"</c>, isolating subclass state from the host's own
|
||||
/// state keys (and from the reserved base-state key). Implementations therefore may use any
|
||||
/// human-readable key (e.g., <c>"next_index"</c>) without worrying about collisions.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="context">A wrapped workflow context that scopes state keys to the
|
||||
/// <see cref="GroupChatManager"/> subclass namespace.</param>
|
||||
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests.
|
||||
/// The default is <see cref="CancellationToken.None"/>.</param>
|
||||
protected virtual ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
=> default;
|
||||
|
||||
/// <summary>
|
||||
/// Invoked when the hosting group chat workflow is being restored from a checkpoint, giving
|
||||
/// subclasses a chance to hydrate any additional state they persisted in
|
||||
/// <see cref="OnCheckpointingAsync"/>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The default implementation is a no-op. Base-class state (currently
|
||||
/// <see cref="IterationCount"/>) is restored automatically by the hosting
|
||||
/// <see cref="Specialized.GroupChatHost"/> before this method is invoked; subclasses do not
|
||||
/// need to call <c>base.OnCheckpointRestoredAsync</c>. The supplied <paramref name="context"/>
|
||||
/// uses the same key-prefixing wrapper as <see cref="OnCheckpointingAsync"/>.
|
||||
/// </remarks>
|
||||
/// <param name="context">A wrapped workflow context that scopes state keys to the
|
||||
/// <see cref="GroupChatManager"/> subclass namespace.</param>
|
||||
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests.
|
||||
/// The default is <see cref="CancellationToken.None"/>.</param>
|
||||
protected virtual ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
=> default;
|
||||
|
||||
// Root checkpoint entry point invoked by the hosting GroupChatHost. Persists the manager's
|
||||
// own base state under the reserved BaseStateKey on the raw context, then delegates to the
|
||||
// subclass-facing OnCheckpointingAsync hook with a wrapped context that prefixes every key
|
||||
// with SubclassStateKeyPrefix.
|
||||
internal async ValueTask CheckpointAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
await context.QueueStateUpdateAsync(BaseStateKey, new GroupChatManagerState(this.IterationCount), cancellationToken: cancellationToken).ConfigureAwait(false);
|
||||
await this.OnCheckpointingAsync(new PrefixingWorkflowContext(context, SubclassStateKeyPrefix), cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
// Root restore entry point invoked by the hosting GroupChatHost. Symmetric to CheckpointAsync.
|
||||
internal async ValueTask RestoreCheckpointAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
GroupChatManagerState? state = await context.ReadStateAsync<GroupChatManagerState>(BaseStateKey, cancellationToken: cancellationToken).ConfigureAwait(false);
|
||||
this.IterationCount = state?.IterationCount ?? 0;
|
||||
await this.OnCheckpointRestoredAsync(new PrefixingWorkflowContext(context, SubclassStateKeyPrefix), cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed record GroupChatManagerState(int IterationCount);
|
||||
|
||||
// IWorkflowContext decorator that prepends a fixed prefix to every state key passed through it.
|
||||
// All non-state members (events, message sending, output yielding, halt requests, trace context,
|
||||
// and runtime characteristics) delegate directly to the wrapped context.
|
||||
internal sealed class PrefixingWorkflowContext(IWorkflowContext inner, string prefix) : IWorkflowContext
|
||||
{
|
||||
private readonly IWorkflowContext _inner = Throw.IfNull(inner);
|
||||
private readonly string _prefix = Throw.IfNullOrEmpty(prefix);
|
||||
|
||||
public IReadOnlyDictionary<string, string>? TraceContext => this._inner.TraceContext;
|
||||
|
||||
public bool ConcurrentRunsEnabled => this._inner.ConcurrentRunsEnabled;
|
||||
|
||||
public ValueTask AddEventAsync(WorkflowEvent workflowEvent, CancellationToken cancellationToken = default)
|
||||
=> this._inner.AddEventAsync(workflowEvent, cancellationToken);
|
||||
|
||||
public ValueTask SendMessageAsync(object message, string? targetId, CancellationToken cancellationToken = default)
|
||||
=> this._inner.SendMessageAsync(message, targetId, cancellationToken);
|
||||
|
||||
public ValueTask YieldOutputAsync(object output, CancellationToken cancellationToken = default)
|
||||
=> this._inner.YieldOutputAsync(output, cancellationToken);
|
||||
|
||||
public ValueTask RequestHaltAsync() => this._inner.RequestHaltAsync();
|
||||
|
||||
public ValueTask<T?> ReadStateAsync<T>(string key, string? scopeName = null, CancellationToken cancellationToken = default)
|
||||
=> this._inner.ReadStateAsync<T>(this.Wrap(key), scopeName, cancellationToken);
|
||||
|
||||
public ValueTask<T> ReadOrInitStateAsync<T>(string key, Func<T> initialStateFactory, string? scopeName = null, CancellationToken cancellationToken = default)
|
||||
=> this._inner.ReadOrInitStateAsync(this.Wrap(key), initialStateFactory, scopeName, cancellationToken);
|
||||
|
||||
public async ValueTask<HashSet<string>> ReadStateKeysAsync(string? scopeName = null, CancellationToken cancellationToken = default)
|
||||
{
|
||||
HashSet<string> rawKeys = await this._inner.ReadStateKeysAsync(scopeName, cancellationToken).ConfigureAwait(false);
|
||||
return [.. rawKeys.Where(k => k.StartsWith(this._prefix, StringComparison.Ordinal))
|
||||
.Select(k => k.Substring(this._prefix.Length))];
|
||||
}
|
||||
|
||||
public ValueTask QueueStateUpdateAsync<T>(string key, T? value, string? scopeName = null, CancellationToken cancellationToken = default)
|
||||
=> this._inner.QueueStateUpdateAsync(this.Wrap(key), value, scopeName, cancellationToken);
|
||||
|
||||
public async ValueTask QueueClearScopeAsync(string? scopeName = null, CancellationToken cancellationToken = default)
|
||||
{
|
||||
// Clearing the entire underlying scope would also remove keys owned by the host and other
|
||||
// subsystems sharing the executor's default scope. Restrict the clear to keys carrying
|
||||
// this wrapper's prefix.
|
||||
HashSet<string> rawKeys = await this._inner.ReadStateKeysAsync(scopeName, cancellationToken).ConfigureAwait(false);
|
||||
foreach (string rawKey in rawKeys)
|
||||
{
|
||||
if (rawKey.StartsWith(this._prefix, StringComparison.Ordinal))
|
||||
{
|
||||
await this._inner.QueueStateUpdateAsync<object>(rawKey, null, scopeName, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private string Wrap(string key) => this._prefix + Throw.IfNullOrEmpty(key);
|
||||
}
|
||||
|
||||
@@ -75,10 +75,14 @@ public sealed class GroupChatWorkflowBuilder
|
||||
{
|
||||
AIAgent[] agents = this._participants.ToArray();
|
||||
|
||||
// GroupChatHost owns the canonical conversation and broadcasts messages directly to every
|
||||
// participant. Participants therefore must not echo their incoming messages back to the host
|
||||
// (which would cause duplicates), but must still reframe other agents' assistant messages as
|
||||
// user messages so each agent's own session reads coherently.
|
||||
AIAgentHostOptions options = new()
|
||||
{
|
||||
ReassignOtherAgentsAsUsers = true,
|
||||
ForwardIncomingMessages = true
|
||||
ForwardIncomingMessages = false
|
||||
};
|
||||
|
||||
Dictionary<AIAgent, ExecutorBinding> agentMap = agents.ToDictionary(a => a, a => a.BindAsExecutor(options));
|
||||
|
||||
@@ -69,4 +69,23 @@ public class RoundRobinGroupChatManager : GroupChatManager
|
||||
base.Reset();
|
||||
this._nextIndex = 0;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
=> context.QueueStateUpdateAsync(StateKey, new RoundRobinGroupChatManagerState(this._nextIndex), cancellationToken: cancellationToken);
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
RoundRobinGroupChatManagerState? state = await context.ReadStateAsync<RoundRobinGroupChatManagerState>(StateKey, cancellationToken: cancellationToken).ConfigureAwait(false);
|
||||
this._nextIndex = state?.NextIndex ?? 0;
|
||||
if (this._nextIndex < 0 || this._nextIndex >= this._agents.Count)
|
||||
{
|
||||
this._nextIndex = 0;
|
||||
}
|
||||
}
|
||||
|
||||
private const string StateKey = "next_index";
|
||||
}
|
||||
|
||||
internal sealed record RoundRobinGroupChatManagerState(int NextIndex);
|
||||
|
||||
@@ -20,12 +20,25 @@ internal sealed class GroupChatHost(
|
||||
AutoSendTurnToken = false
|
||||
};
|
||||
|
||||
private const string HistoryStateKey = nameof(_history);
|
||||
private const string CurrentSpeakerStateKey = nameof(_currentSpeakerExecutorId);
|
||||
|
||||
private readonly AIAgent[] _agents = agents;
|
||||
private readonly Dictionary<AIAgent, ExecutorBinding> _agentMap = agentMap;
|
||||
private readonly Func<IReadOnlyList<AIAgent>, GroupChatManager> _managerFactory = managerFactory;
|
||||
|
||||
private GroupChatManager? _manager;
|
||||
|
||||
// Canonical conversation accumulated across turns. Each participant maintains its own per-agent
|
||||
// session/thread; the host keeps this only as the source of truth for the manager hooks
|
||||
// (SelectNextAgentAsync / ShouldTerminateAsync) and for the workflow's final output.
|
||||
private List<ChatMessage> _history = [];
|
||||
|
||||
// Executor id of the participant we most recently dispatched a TurnToken to – i.e., the current
|
||||
// speaker whose response is about to arrive. Used to exclude that participant from the next
|
||||
// broadcast (its own session already contains the message it produced).
|
||||
private string? _currentSpeakerExecutorId;
|
||||
|
||||
protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)
|
||||
=> base.ConfigureProtocol(protocolBuilder).YieldsOutput<List<ChatMessage>>();
|
||||
|
||||
@@ -33,30 +46,105 @@ internal sealed class GroupChatHost(
|
||||
{
|
||||
this._manager ??= this._managerFactory(this._agents);
|
||||
|
||||
if (!await this._manager.ShouldTerminateAsync(messages, cancellationToken).ConfigureAwait(false))
|
||||
// The delta arriving here is either the initial user input (turn 0) or the most recent speaker's
|
||||
// response (subsequent turns) – participants no longer echo incoming messages back to the host.
|
||||
if (messages.Count > 0)
|
||||
{
|
||||
var filtered = await this._manager.UpdateHistoryAsync(messages, cancellationToken).ConfigureAwait(false);
|
||||
messages = filtered is null || ReferenceEquals(filtered, messages) ? messages : [.. filtered];
|
||||
this._history.AddRange(messages);
|
||||
}
|
||||
|
||||
if (await this._manager.SelectNextAgentAsync(messages, cancellationToken).ConfigureAwait(false) is AIAgent nextAgent &&
|
||||
this._agentMap.TryGetValue(nextAgent, out var executor))
|
||||
if (await this._manager.ShouldTerminateAsync(this._history, cancellationToken).ConfigureAwait(false))
|
||||
{
|
||||
await this.CompleteAsync(context, cancellationToken).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
|
||||
if (messages.Count > 0)
|
||||
{
|
||||
IEnumerable<ChatMessage> filteredDelta = await this._manager.UpdateHistoryAsync(messages, cancellationToken).ConfigureAwait(false);
|
||||
List<ChatMessage> broadcastMessages = filteredDelta is null
|
||||
? messages
|
||||
: (ReferenceEquals(filteredDelta, messages) ? messages : [.. filteredDelta]);
|
||||
|
||||
if (broadcastMessages.Count > 0)
|
||||
{
|
||||
this._manager.IterationCount++;
|
||||
await context.SendMessageAsync(messages, executor.Id, cancellationToken).ConfigureAwait(false);
|
||||
await context.SendMessageAsync(new TurnToken(emitEvents), executor.Id, cancellationToken).ConfigureAwait(false);
|
||||
return;
|
||||
await this.BroadcastAsync(broadcastMessages, context, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
this._manager = null;
|
||||
await context.YieldOutputAsync(messages, cancellationToken).ConfigureAwait(false);
|
||||
if (await this._manager.SelectNextAgentAsync(this._history, cancellationToken).ConfigureAwait(false) is AIAgent nextAgent &&
|
||||
this._agentMap.TryGetValue(nextAgent, out ExecutorBinding? executor))
|
||||
{
|
||||
this._manager.IterationCount++;
|
||||
this._currentSpeakerExecutorId = executor.Id;
|
||||
await context.SendMessageAsync(new TurnToken(emitEvents), executor.Id, cancellationToken).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
|
||||
await this.CompleteAsync(context, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private ValueTask BroadcastAsync(List<ChatMessage> messages, IWorkflowContext context, CancellationToken cancellationToken)
|
||||
{
|
||||
List<Task>? sendTasks = null;
|
||||
foreach (ExecutorBinding participant in this._agentMap.Values)
|
||||
{
|
||||
if (string.Equals(participant.Id, this._currentSpeakerExecutorId, StringComparison.Ordinal))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
(sendTasks ??= []).Add(context.SendMessageAsync(messages, participant.Id, cancellationToken).AsTask());
|
||||
}
|
||||
|
||||
return sendTasks is null ? default : new ValueTask(Task.WhenAll(sendTasks));
|
||||
}
|
||||
|
||||
private async ValueTask CompleteAsync(IWorkflowContext context, CancellationToken cancellationToken)
|
||||
{
|
||||
List<ChatMessage> output = this._history;
|
||||
this._history = [];
|
||||
this._currentSpeakerExecutorId = null;
|
||||
this._manager = null;
|
||||
|
||||
await context.YieldOutputAsync(output, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
protected override ValueTask ResetAsync()
|
||||
{
|
||||
this._manager = null;
|
||||
this._history = [];
|
||||
this._currentSpeakerExecutorId = null;
|
||||
|
||||
return base.ResetAsync();
|
||||
}
|
||||
|
||||
ValueTask IResettableExecutor.ResetAsync() => this.ResetAsync();
|
||||
|
||||
protected internal override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
Task historyTask = context.QueueStateUpdateAsync(HistoryStateKey, this._history, cancellationToken: cancellationToken).AsTask();
|
||||
Task currentSpeakerTask = context.QueueStateUpdateAsync(CurrentSpeakerStateKey, this._currentSpeakerExecutorId, cancellationToken: cancellationToken).AsTask();
|
||||
Task baseTask = base.OnCheckpointingAsync(context, cancellationToken).AsTask();
|
||||
|
||||
// Eagerly materialize the manager so subclass state (e.g., the round-robin cursor) gets
|
||||
// persisted on every checkpoint, even if no turn has been taken yet since the host was constructed.
|
||||
this._manager ??= this._managerFactory(this._agents);
|
||||
Task managerTask = this._manager.CheckpointAsync(context, cancellationToken).AsTask();
|
||||
|
||||
await Task.WhenAll(historyTask, currentSpeakerTask, baseTask, managerTask).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
this._history = await context.ReadStateAsync<List<ChatMessage>>(HistoryStateKey, cancellationToken: cancellationToken).ConfigureAwait(false) ?? [];
|
||||
this._currentSpeakerExecutorId = await context.ReadStateAsync<string?>(CurrentSpeakerStateKey, cancellationToken: cancellationToken).ConfigureAwait(false);
|
||||
|
||||
// Instantiate the manager eagerly so its restore hook can rehydrate IterationCount and any
|
||||
// subclass-defined state (e.g., RoundRobinGroupChatManager._nextIndex).
|
||||
this._manager = this._managerFactory(this._agents);
|
||||
await this._manager.RestoreCheckpointAsync(context, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
await base.OnCheckpointRestoredAsync(context, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -101,6 +101,8 @@ internal static partial class WorkflowsJsonUtilities
|
||||
[JsonSerializable(typeof(MagenticPlanReviewRequest))]
|
||||
[JsonSerializable(typeof(MagenticPlanReviewResponse))]
|
||||
[JsonSerializable(typeof(MagenticTaskState))]
|
||||
[JsonSerializable(typeof(GroupChatManagerState))]
|
||||
[JsonSerializable(typeof(RoundRobinGroupChatManagerState))]
|
||||
[JsonSerializable(typeof(ResetChatSignal))]
|
||||
|
||||
// Event Types
|
||||
|
||||
Reference in New Issue
Block a user