agent/agent-framework
2
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

.NET: Add AgentWorkflowBuilder group chat (#861)

Browse Source 
* Add AgentWorkflowBuilder group chat

And fix a variety of issues along the way:
- Use DateTime{Offset}.UtcNow rather than Now
- AIAgentHostExecutor shouldn't be publishing empty messages
- Sequential workflows should be flowing all history and not just the output from the previous agent as the input into the next agent
- Renamed some of the new agent workflow methods... still not super happy with the shape, though
- Simplified handoffs builder, e.g. using a hashset with a custom comparer instead of a dictionary
- Improved multi-service use by trying to change assistant->user role for messages created by other agents
- Changed MessageMerger to rely on M.E.AI's coalescing more and to avoid empty contents / text
- Ensured that messages from ChatClientAgent include MessageId and CreatedAt timestamps
- Avoided including instructions for agents in a handoff workflow that don't have any handoffs
- Removed the unnecessary end function in handoffs
- Improved naming of executors to include agent name for debuggability
- Use "N" formatting with Guid.ToString everywhere, to avoid the unnecessary extra dash character which is also not valid in various places (like function tool names)
- Replace `params T[]` with `params IEnumerable<T>` to make public APIs more flexible in what they consume

* Address feedback

- Fix unintentional provider change in sample
This commit is contained in:
Stephen Toub
2025-09-24 16:38:34 +00:00
committed by GitHub
parent 7c70c33157
commit 03ef7f054f
51 changed files with 888 additions and 619 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/06_GroupChat_Workflow.cs
+5 -224
View File
@@ -17,16 +17,11 @@ namespace Microsoft.Agents.Workflows.Sample;
internal static class Step6EntryPoint
{
public static Workflow<List<ChatMessage>> CreateWorkflow(int maxTurns)
{
GroupChatBuilder builder =
GroupChatBuilder.Create<RoundRobinGroupChatManager, RoundRobinGroupChatManagerOptions>
(options => options.MaxTurns = maxTurns)
.AddParticipant(new HelloAgent(), shouldEmitEvents: true)
.AddParticipant(new EchoAgent(), shouldEmitEvents: true);
return builder.ReduceToWorkflow();
}
public static Workflow<List<ChatMessage>> CreateWorkflow(int maxTurns) =>
AgentWorkflowBuilder
.CreateGroupChatBuilderWith(agents => new AgentWorkflowBuilder.RoundRobinGroupChatManager(agents) { MaximumIterationCount = maxTurns })
.AddParticipants(new HelloAgent(), new EchoAgent())
.Build();
public static async ValueTask RunAsync(TextWriter writer, int maxSteps = 2)
{
@@ -53,39 +48,6 @@ internal static class Step6EntryPoint
}
}
}
private sealed class RoundRobinGroupChatManagerOptions : GroupChatManagerOptions
{
public int? MaxTurns { get; set; }
}
private sealed class RoundRobinGroupChatManager() : GroupChatManager<RoundRobinGroupChatManagerOptions>
{
public int TurnCount { get; private set; }
public int? MaxTurns { get; private set; }
protected internal override void Configure(RoundRobinGroupChatManagerOptions options)
{
base.Configure(options);
this.MaxTurns = options.MaxTurns;
}
public override int? GetNextTurnExecutor(GroupChatHistory history)
{
if (this.ParticipantIds.Length == 0)
{
throw new InvalidOperationException("No participants in the group chat.");
}
if (this.TurnCount >= this.MaxTurns)
{
return null;
}
return this.TurnCount++ % this.ParticipantIds.Length;
}
}
}
internal sealed class HelloAgent(string id = nameof(HelloAgent)) : AIAgent
@@ -175,184 +137,3 @@ internal sealed class EchoAgent(string id = nameof(EchoAgent)) : AIAgent
}
internal sealed class EchoAgentThread() : InMemoryAgentThread();
internal sealed class GroupChatHistory
{
private readonly List<ChatMessage> _messages = [];
private int _bookmark;
public void AddMessage(ChatMessage message) =>
this._messages.Add(message);
public void AddMessages(IEnumerable<ChatMessage> messages) =>
this._messages.AddRange(messages);
public void UpdateBookmark() =>
this._bookmark = this._messages.Count;
public IReadOnlyList<ChatMessage> FullHistory => this._messages.AsReadOnly();
public IEnumerable<ChatMessage> NewMessagesThisTurn => this._messages.Skip(this._bookmark);
}
internal class GroupChatManagerOptions;
internal abstract class GroupChatManager
{
public string[] ParticipantIds { get; internal init; } = [];
public abstract int? GetNextTurnExecutor(GroupChatHistory history);
}
internal abstract class GroupChatManager<TOptions> : GroupChatManager where TOptions : GroupChatManagerOptions, new()
{
protected internal virtual void Configure(TOptions options) { }
}
internal sealed class GroupChatBuilder
{
private readonly List<ExecutorIsh> _participants = [];
private readonly List<bool> _shouldEmitEvents = [];
private readonly Func<string[], GroupChatManager> _managerFactory;
private GroupChatBuilder(Func<string[], GroupChatManager> managerFactory)
{
this._managerFactory = managerFactory;
}
public static GroupChatBuilder Create<TManager>() where TManager : GroupChatManager, new() =>
new(participantIds => new TManager() { ParticipantIds = participantIds });
public static GroupChatBuilder Create<TManager, TOptions>(Action<TOptions> configure)
where TManager : GroupChatManager<TOptions>, new()
where TOptions : GroupChatManagerOptions, new()
{
TOptions options = new();
configure(options);
return new GroupChatBuilder(participantIds =>
{
TManager manager = new() { ParticipantIds = participantIds };
manager.Configure(options);
return manager;
});
}
public GroupChatBuilder AddParticipant(ExecutorIsh executor, bool shouldEmitEvents = false)
{
this._participants.Add(executor);
this._shouldEmitEvents.Add(shouldEmitEvents);
return this;
}
public GroupChatBuilder AddParticipants(params ExecutorIsh[] executors)
{
this._participants.AddRange(executors);
return this;
}
public Workflow<List<ChatMessage>> ReduceToWorkflow()
{
string[] participantIds = this._participants.Select(identified => identified.Id).ToArray();
GroupChatHost host = new(this._shouldEmitEvents.ToArray(), this._managerFactory(participantIds));
WorkflowBuilder builder = new WorkflowBuilder(host)
.AddFanOutEdge(host, targets: this._participants.ToArray());
foreach (ExecutorIsh participant in this._participants)
{
builder.AddEdge(participant, host);
}
return builder.Build<List<ChatMessage>>();
//bool IsMessageType(object? message) => message is ChatMessage || message is IEnumerable<ChatMessage>;
}
private sealed class TurnAssignedEvent(string executorId, string nextSpeakerId) : ExecutorEvent(executorId, data: nextSpeakerId);
private sealed class GroupChatHost : Executor
{
private readonly bool[] _shouldEmitEvents;
private readonly GroupChatManager _manager;
private readonly bool _autoStartConversation;
private readonly GroupChatHistory _history = new();
public GroupChatHost(bool[] shouldEmitEvents, GroupChatManager manager, bool autoStartConversation = false) : base(nameof(GroupChatHost))
{
this._shouldEmitEvents = shouldEmitEvents;
this._manager = manager ?? throw new ArgumentNullException(nameof(manager));
this._autoStartConversation = autoStartConversation;
}
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) =>
routeBuilder.AddHandler<List<ChatMessage>>(this.HandleChatMessagesAsync)
.AddHandler<ChatMessage>(this.HandleChatMessageAsync)
.AddHandler<TurnToken>(this.AssignNextTurnAsync);
private async Task TryAutoStartConversationAsync(IWorkflowContext context)
{
if (this._autoStartConversation && this.TryEnterConversation())
{
await this.AssignNextTurnAsync(new TurnToken(emitEvents: false), context).ConfigureAwait(false);
}
}
private async ValueTask HandleChatMessagesAsync(List<ChatMessage> initialMessages, IWorkflowContext context)
{
this._history.AddMessages(initialMessages);
await context.SendMessageAsync(initialMessages).ConfigureAwait(false);
await this.TryAutoStartConversationAsync(context).ConfigureAwait(false);
}
private async ValueTask HandleChatMessageAsync(ChatMessage message, IWorkflowContext context)
{
// First, add the message to the history, then forward to all executors
this._history.AddMessage(message);
await context.SendMessageAsync(message).ConfigureAwait(false);
await this.TryAutoStartConversationAsync(context).ConfigureAwait(false);
}
private int _inConversationFlag;
/// <summary>
/// Atomically switches to "in conversation" state if not already in that state.
/// </summary>
/// <returns><see langword="true"/> if the state was changed, <see langword="false"/> otherwise.</returns>
private bool TryEnterConversation() =>
Interlocked.CompareExchange(ref this._inConversationFlag, 1, 0) == 0;
private bool _shouldHostEmitEvents;
private async ValueTask AssignNextTurnAsync(TurnToken token, IWorkflowContext context)
{
if (this.TryEnterConversation())
{
// Capture the initial turn token's EmitEvents setting
this._shouldHostEmitEvents = token.EmitEvents ?? false;
}
int? nextSpeakerIndex = this._manager.GetNextTurnExecutor(this._history);
if (nextSpeakerIndex is null)
{
await context.AddEventAsync(new WorkflowCompletedEvent())
.ConfigureAwait(false);
return;
}
string nextSpeakerId = this._manager.ParticipantIds[nextSpeakerIndex.Value];
if (this._shouldHostEmitEvents)
{
await context.AddEventAsync(new TurnAssignedEvent(this.Id, nextSpeakerId))
.ConfigureAwait(false);
}
await context.SendMessageAsync(new TurnToken(this._shouldEmitEvents[nextSpeakerIndex.Value]), nextSpeakerId)
.ConfigureAwait(false);
}
}
}