From 3baf5279098fabf5009c1b60d4f9fd40b6b93048 Mon Sep 17 00:00:00 2001 From: Jacob Alber Date: Fri, 22 May 2026 14:49:42 -0400 Subject: [PATCH] feat: SequentialWorkflowBuilder and ConcurrentWorkflowBuilder, OrchestrationBuilderBase Promotes the Sequential and Concurrent orchestration shapes to first-class fluent builder classes, matching Handoff / GroupChat / Magentic. Users can call `WithOutputFrom(agents)` / `WithIntermediateOutputFrom(agents)` to control which agents are designated output / intermediate sources; when no designation call is made, the Python-aligned defaults apply (terminal aggregator output + every agent intermediate; Concurrent also tags per-agent accumulators). `AgentWorkflowBuilder.BuildSequential(...)` and `BuildConcurrent(...)` are kept and now delegate to the new builders; observable behavior unchanged. Five static factories now mirror each other: - `AgentWorkflowBuilder.CreateSequentialBuilderWith(params IEnumerable)` - `AgentWorkflowBuilder.CreateConcurrentBuilderWith(params IEnumerable)` - `AgentWorkflowBuilder.CreateHandoffBuilderWith(AIAgent)` (already existed) - `AgentWorkflowBuilder.CreateGroupChatBuilderWith(Func<...>)` (already existed) - `AgentWorkflowBuilder.CreateMagenticBuilderWith(AIAgent)` (new) OrchestrationBuilderBase ------------------------ New abstract `OrchestrationBuilderBase` unifies the shared fluent surface across all five orchestration builders: `WithName`, `WithDescription`, `WithOutputFrom`, `WithIntermediateOutputFrom`, and the `ApplyOutputDesignations(builder, agentMap, kind, applyDefaults)` helper that either replays the user's designations or invokes the orchestration-specific defaults. Removes ~150 LOC of duplicated designation-management code from the four non-Handoff builders, plus the equivalent from `HandoffWorkflowBuilderCore`. Tests ----- - New `SequentialWorkflowBuilderTests.cs` / `ConcurrentWorkflowBuilderTests.cs` (replace the old `AgentWorkflowBuilder.{Sequential,Concurrent}Tests.cs` nested-class files). Method names normalized to `Test__[Async]`. - Shared helpers (`DoubleEchoAgent`, `DoubleEchoAgentWithBarrier`, `WorkflowRunResult`, `RunWorkflow*`) moved from the old `AgentWorkflowBuilderTests` partial class into a new `OrchestrationTestHelpers` static class in `OrchestrationTestHelpers.cs`. Downstream test files (Group Chat, Handoff, Sequential, Concurrent) updated to qualify with `OrchestrationTestHelpers.*`. - A new `AgentWorkflowBuilderTests.cs` covers the static surface directly: `BuildSequential` / `BuildConcurrent` invariants and aggregator wiring, plus null-rejection + round-trip checks for every `Create*BuilderWith` factory. - New AsAgent intermediate-suppression tests on a nested `AsAgentForwarding` class for each of Sequential and Concurrent: build with only the terminal agent designated via `WithOutputFrom`, run via `AsAIAgent(...)`, assert via `AgentResponseUpdate.AuthorName` that intermediate agents do not surface. Both join the `FuturesSerial` collection. - New `Test__WithDescriptionPropagatesToWorkflow` smoke tests on Sequential and Concurrent (newly available via the base class). 625/625 unit tests pass on net10.0. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../AgentWorkflowBuilder.cs | 97 +++---- .../ConcurrentWorkflowBuilder.cs | 104 +++++++ .../GroupChatWorkflowBuilder.cs | 121 +-------- .../HandoffWorkflowBuilder.cs | 127 +-------- .../MagenticWorkflowBuilder.cs | 116 +------- .../OrchestrationBuilderBase.cs | 154 +++++++++++ .../SequentialWorkflowBuilder.cs | 85 ++++++ .../AgentWorkflowBuilder.ConcurrentTests.cs | 77 ------ .../AgentWorkflowBuilder.SequentialTests.cs | 90 ------- .../AgentWorkflowBuilderTests.cs | 254 ++++++++++-------- .../ConcurrentWorkflowBuilderTests.cs | 165 ++++++++++++ .../GroupChatWorkflowBuilderTests.cs | 34 +-- .../HandoffWorkflowBuilderTests.cs | 14 +- .../OrchestrationTestHelpers.cs | 131 +++++++++ .../SequentialWorkflowBuilderTests.cs | 184 +++++++++++++ 15 files changed, 1057 insertions(+), 696 deletions(-) create mode 100644 dotnet/src/Microsoft.Agents.AI.Workflows/ConcurrentWorkflowBuilder.cs create mode 100644 dotnet/src/Microsoft.Agents.AI.Workflows/OrchestrationBuilderBase.cs create mode 100644 dotnet/src/Microsoft.Agents.AI.Workflows/SequentialWorkflowBuilder.cs delete mode 100644 dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilder.ConcurrentTests.cs delete mode 100644 dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilder.SequentialTests.cs create mode 100644 dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ConcurrentWorkflowBuilderTests.cs create mode 100644 dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/OrchestrationTestHelpers.cs create mode 100644 dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/SequentialWorkflowBuilderTests.cs diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs index 20eb6d903b..84fc9e8910 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs @@ -3,9 +3,7 @@ using System; using System.Collections.Generic; using System.Diagnostics.CodeAnalysis; -using System.Linq; using System.Threading.Tasks; -using Microsoft.Agents.AI.Workflows.Specialized; using Microsoft.Extensions.AI; using Microsoft.Shared.Diagnostics; @@ -37,33 +35,10 @@ public static partial class AgentWorkflowBuilder { Throw.IfNullOrEmpty(agents); - // Create a builder that chains the agents together in sequence. The workflow simply begins - // with the first agent in the sequence. - - AIAgentHostOptions options = new() - { - ReassignOtherAgentsAsUsers = true, - ForwardIncomingMessages = true, - }; - - List agentExecutors = agents.Select(agent => agent.BindAsExecutor(options)).ToList(); - - ExecutorBinding previous = agentExecutors[0]; - WorkflowBuilder builder = new(previous); - - foreach (ExecutorBinding next in agentExecutors.Skip(1)) - { - builder.AddEdge(previous, next); - previous = next; - } - - OutputMessagesExecutor end = new(); - builder = builder.AddEdge(previous, end) - .WithOutputFrom(end) - .WithIntermediateOutputFrom(agentExecutors); + SequentialWorkflowBuilder builder = new(agents); if (workflowName is not null) { - builder = builder.WithName(workflowName); + builder.WithName(workflowName); } return builder.Build(); } @@ -109,42 +84,14 @@ public static partial class AgentWorkflowBuilder { Throw.IfNull(agents); - // A workflow needs a starting executor, so we create one that forwards everything to each agent. - ChatForwardingExecutor start = new("Start"); - WorkflowBuilder builder = new(start); - - // For each agent, we create an executor to host it and an accumulator to batch up its output messages, - // so that the final accumulator receives a single list of messages from each agent. Otherwise, the - // accumulator would not be able to determine what came from what agent, as there's currently no - // provenance tracking exposed in the workflow context passed to a handler. - - ExecutorBinding[] agentExecutors = (from agent in agents - select agent.BindAsExecutor(new AIAgentHostOptions() { ReassignOtherAgentsAsUsers = true })).ToArray(); - ExecutorBinding[] accumulators = [.. from agent in agentExecutors select (ExecutorBinding)new AggregateTurnMessagesExecutor($"Batcher/{agent.Id}")]; - builder.AddFanOutEdge(start, agentExecutors); - - for (int i = 0; i < agentExecutors.Length; i++) - { - builder.AddEdge(agentExecutors[i], accumulators[i]); - } - - // Create the accumulating executor that will gather the results from each agent, and connect - // each agent's accumulator to it. If no aggregation function was provided, we default to returning - // the last message from each agent - aggregator ??= static lists => (from list in lists where list.Count > 0 select list.Last()).ToList(); - - Func> endFactory = - (_, __) => new(new ConcurrentEndExecutor(agentExecutors.Length, aggregator)); - - ExecutorBinding end = endFactory.BindExecutor(ConcurrentEndExecutor.ExecutorId); - - builder.AddFanInBarrierEdge(accumulators, end); - - builder = builder.WithOutputFrom(end) - .WithIntermediateOutputFrom([.. agentExecutors, .. accumulators]); + ConcurrentWorkflowBuilder builder = new(agents); if (workflowName is not null) { - builder = builder.WithName(workflowName); + builder.WithName(workflowName); + } + if (aggregator is not null) + { + builder.WithAggregator(aggregator); } return builder.Build(); } @@ -182,4 +129,32 @@ public static partial class AgentWorkflowBuilder Throw.IfNull(managerFactory); return new GroupChatWorkflowBuilder(managerFactory); } + + /// Creates a new with the given pipeline of . + /// The sequence of agents to compose into a sequential workflow. + /// The builder for creating a sequential workflow. + public static SequentialWorkflowBuilder CreateSequentialBuilderWith(params IEnumerable agents) + { + Throw.IfNull(agents); + return new SequentialWorkflowBuilder(agents); + } + + /// Creates a new with the given participating . + /// The set of agents to compose into a concurrent workflow. + /// The builder for creating a concurrent workflow. + public static ConcurrentWorkflowBuilder CreateConcurrentBuilderWith(params IEnumerable agents) + { + Throw.IfNull(agents); + return new ConcurrentWorkflowBuilder(agents); + } + + /// Creates a new with the given . + /// The LLM-powered manager agent that coordinates the team. + /// The builder for creating a Magentic workflow. + [Experimental(DiagnosticConstants.ExperimentalFeatureDiagnostic)] + public static MagenticWorkflowBuilder CreateMagenticBuilderWith(AIAgent managerAgent) + { + Throw.IfNull(managerAgent); + return new MagenticWorkflowBuilder(managerAgent); + } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/ConcurrentWorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/ConcurrentWorkflowBuilder.cs new file mode 100644 index 0000000000..1ac8a9c9fd --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/ConcurrentWorkflowBuilder.cs @@ -0,0 +1,104 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows.Specialized; +using Microsoft.Extensions.AI; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Agents.AI.Workflows; + +/// +/// Fluent builder for concurrent agent workflows: a fan-out start that broadcasts the +/// incoming messages to every participating agent, a per-agent accumulator that batches +/// each agent's outgoing messages, and a fan-in aggregator that reduces them into a +/// single output list. +/// +/// +/// When no explicit output designations are made, the default is the Python-aligned +/// shape: the terminal aggregator is the workflow output, and every participating agent +/// (plus its per-agent accumulator) is designated as an intermediate output source. +/// Calling +/// or +/// at all suppresses these defaults. +/// +public sealed class ConcurrentWorkflowBuilder : OrchestrationBuilderBase +{ + private readonly List _agents = []; + private Func>, List>? _aggregator; + + /// + /// Initializes a new with the given participating + /// . + /// + public ConcurrentWorkflowBuilder(params IEnumerable agents) + { + Throw.IfNull(agents); + foreach (AIAgent agent in agents) + { + Throw.IfNull(agent, nameof(agents)); + this._agents.Add(agent); + } + } + + /// + /// Sets the aggregator function. If not called, defaults to returning the last message + /// from each agent that produced at least one message. + /// + public ConcurrentWorkflowBuilder WithAggregator(Func>, List> aggregator) + { + this._aggregator = Throw.IfNull(aggregator); + return this; + } + + /// Builds the configured concurrent workflow. + public Workflow Build() + { + if (this._agents.Count == 0) + { + throw new ArgumentException("At least one agent must be provided to the ConcurrentWorkflowBuilder.", "agents"); + } + + ChatForwardingExecutor start = new("Start"); + WorkflowBuilder builder = new(start); + + Dictionary agentMap = new(AIAgentIDEqualityComparer.Instance); + ExecutorBinding[] agentExecutors = new ExecutorBinding[this._agents.Count]; + ExecutorBinding[] accumulators = new ExecutorBinding[this._agents.Count]; + AIAgentHostOptions options = new() { ReassignOtherAgentsAsUsers = true }; + for (int i = 0; i < this._agents.Count; i++) + { + AIAgent agent = this._agents[i]; + ExecutorBinding binding = agent.BindAsExecutor(options); + agentExecutors[i] = binding; + agentMap[agent] = binding; + accumulators[i] = new AggregateTurnMessagesExecutor($"Batcher/{binding.Id}"); + } + + builder.AddFanOutEdge(start, agentExecutors); + for (int i = 0; i < agentExecutors.Length; i++) + { + builder.AddEdge(agentExecutors[i], accumulators[i]); + } + + Func>, List> aggregator = + this._aggregator ?? (static lists => (from list in lists where list.Count > 0 select list.Last()).ToList()); + + Func> endFactory = + (_, __) => new(new ConcurrentEndExecutor(agentExecutors.Length, aggregator)); + + ExecutorBinding end = endFactory.BindExecutor(ConcurrentEndExecutor.ExecutorId); + builder.AddFanInBarrierEdge(accumulators, end); + + this.ApplyMetadata(builder); + this.ApplyOutputDesignations(builder, agentMap, "concurrent", () => + { + builder.WithOutputFrom(end); + builder.WithIntermediateOutputFrom([.. agentExecutors, .. accumulators]); + }); + + return builder.Build(); + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/GroupChatWorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/GroupChatWorkflowBuilder.cs index a19915f5d5..2f587f32b7 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/GroupChatWorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/GroupChatWorkflowBuilder.cs @@ -1,4 +1,4 @@ -// Copyright (c) Microsoft. All rights reserved. +// Copyright (c) Microsoft. All rights reserved. using System; using System.Collections.Generic; @@ -12,14 +12,10 @@ namespace Microsoft.Agents.AI.Workflows; /// /// Provides a builder for specifying group chat relationships between agents and building the resulting workflow. /// -public sealed class GroupChatWorkflowBuilder +public sealed class GroupChatWorkflowBuilder : OrchestrationBuilderBase { private readonly Func, GroupChatManager> _managerFactory; private readonly HashSet _participants = new(AIAgentIDEqualityComparer.Instance); - private string _name = string.Empty; - private string _description = string.Empty; - - private Dictionary>? _outputDesignations; internal GroupChatWorkflowBuilder(Func, GroupChatManager> managerFactory) => this._managerFactory = managerFactory; @@ -46,70 +42,6 @@ public sealed class GroupChatWorkflowBuilder return this; } - /// - /// Sets the human-readable name for the workflow. - /// - /// The name of the workflow. - /// This instance of the . - public GroupChatWorkflowBuilder WithName(string name) - { - this._name = name; - return this; - } - - /// - /// Sets the description for the workflow. - /// - /// The description of what the workflow does. - /// This instance of the . - public GroupChatWorkflowBuilder WithDescription(string description) - { - this._description = description; - return this; - } - - /// - /// Designates the given as sources of terminal workflow output. - /// Calling any output-designation method (this or ) - /// suppresses the orchestration-specific defaults: only the user-specified designations - /// reach the inner . - /// - public GroupChatWorkflowBuilder WithOutputFrom(params IEnumerable agents) - { - Throw.IfNull(agents); - this._outputDesignations ??= new(AIAgentIDEqualityComparer.Instance); - foreach (AIAgent agent in agents) - { - Throw.IfNull(agent, nameof(agents)); - if (!this._outputDesignations.ContainsKey(agent)) - { - this._outputDesignations[agent] = []; - } - } - return this; - } - - /// - /// Designates the given as sources of intermediate workflow output. - /// See for the defaults-suppression semantics. - /// - public GroupChatWorkflowBuilder WithIntermediateOutputFrom(IEnumerable agents) - { - Throw.IfNull(agents); - this._outputDesignations ??= new(AIAgentIDEqualityComparer.Instance); - foreach (AIAgent agent in agents) - { - Throw.IfNull(agent, nameof(agents)); - if (!this._outputDesignations.TryGetValue(agent, out HashSet? tags)) - { - tags = []; - this._outputDesignations[agent] = tags; - } - tags.Add(OutputTag.Intermediate); - } - return this; - } - /// /// Builds a composed of agents that operate via group chat, with the next /// agent to process messages selected by the group chat manager. @@ -137,15 +69,7 @@ public sealed class GroupChatWorkflowBuilder ExecutorBinding host = groupChatHostFactory.BindExecutor(nameof(GroupChatHost)); WorkflowBuilder builder = new(host); - if (!string.IsNullOrEmpty(this._name)) - { - builder = builder.WithName(this._name); - } - - if (!string.IsNullOrEmpty(this._description)) - { - builder = builder.WithDescription(this._description); - } + this.ApplyMetadata(builder); foreach (var participant in agentMap.Values) { @@ -154,48 +78,15 @@ public sealed class GroupChatWorkflowBuilder .AddEdge(participant, host); } - this.ApplyOutputDesignations(builder, host, agentMap); - return builder.Build(); - } - - private void ApplyOutputDesignations( - WorkflowBuilder builder, - ExecutorBinding host, - Dictionary agentMap) - { - if (this._outputDesignations is null) + this.ApplyOutputDesignations(builder, agentMap, "group chat", () => { - // Defaults (matches Python group-chat orchestration): - // host -> terminal output - // participants-> intermediate output builder.WithOutputFrom(host); if (agentMap.Count > 0) { builder.WithIntermediateOutputFrom([.. agentMap.Values]); } - return; - } + }); - foreach (AIAgent agent in this._outputDesignations.Keys) - { - if (!agentMap.TryGetValue(agent, out ExecutorBinding? binding)) - { - throw new InvalidOperationException( - $"Output designation references agent '{agent.Name ?? agent.Id}', which is not a participant in this group chat workflow."); - } - - HashSet tags = this._outputDesignations[agent]; - if (tags.Count == 0) - { - builder.WithOutputFrom(binding); - } - else - { - foreach (OutputTag tag in tags) - { - builder.WithOutputFrom(binding, tag); - } - } - } + return builder.Build(); } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/HandoffWorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/HandoffWorkflowBuilder.cs index 26e83c84f5..b443b07ff5 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/HandoffWorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/HandoffWorkflowBuilder.cs @@ -39,7 +39,8 @@ public sealed class HandoffWorkflowBuilder(AIAgent initialAgent) : HandoffWorkfl /// Provides a builder for specifying the handoff relationships between agents and building the resulting workflow. /// [Experimental(DiagnosticConstants.ExperimentalFeatureDiagnostic)] -public class HandoffWorkflowBuilderCore where TBuilder : HandoffWorkflowBuilderCore +public class HandoffWorkflowBuilderCore : OrchestrationBuilderBase + where TBuilder : HandoffWorkflowBuilderCore { /// /// The prefix for function calls that trigger handoffs to other agents; the full name is then `{FunctionPrefix}<agent_id>`, @@ -55,8 +56,6 @@ public class HandoffWorkflowBuilderCore where TBuilder : HandoffWorkfl private bool _emitAgentResponseUpdateEvents; private HandoffToolCallFilteringBehavior _toolCallFilteringBehavior = HandoffToolCallFilteringBehavior.HandoffOnly; private bool _returnToPrevious; - private string? _name; - private string? _description; // Autonomous mode configuration. When enabled, an agent's response that doesn't include a // handoff triggers another invocation of that same agent with the continuation prompt, up to @@ -74,16 +73,6 @@ public class HandoffWorkflowBuilderCore where TBuilder : HandoffWorkfl // if true, the workflow ends (and the autonomous loop, if any, terminates). private Func, ValueTask>? _terminationCondition; - /// - /// Memoized output designations. means the user has not made any - /// explicit designation, and the orchestration-specific defaults will be applied at - /// time. A non-null (possibly empty) dictionary means the user took - /// control and only these designations will be replayed onto the inner - /// . An entry's value is the set of tags requested for the - /// agent — an empty set encodes a terminal-only designation. - /// - private Dictionary>? _outputDesignations; - /// /// Initializes a new instance of the class with no handoff relationships. /// @@ -126,20 +115,6 @@ public class HandoffWorkflowBuilderCore where TBuilder : HandoffWorkfl return (TBuilder)this; } - /// - public TBuilder WithName(string name) - { - this._name = name; - return (TBuilder)this; - } - - /// - public TBuilder WithDescription(string description) - { - this._description = description; - return (TBuilder)this; - } - /// /// Sets a value indicating whether agent streaming update events should be emitted during execution. /// If , the value will be taken from the @@ -185,48 +160,6 @@ public class HandoffWorkflowBuilderCore where TBuilder : HandoffWorkfl return (TBuilder)this; } - /// - /// Designates the given as sources of terminal workflow output. - /// Calling any output-designation method (this or ) - /// suppresses the orchestration-specific defaults: only the user-specified designations - /// reach the inner . To restore defaults, build a fresh builder. - /// - public TBuilder WithOutputFrom(params IEnumerable agents) - { - Throw.IfNull(agents); - this._outputDesignations ??= new(AIAgentIDEqualityComparer.Instance); - foreach (AIAgent agent in agents) - { - Throw.IfNull(agent, nameof(agents)); - if (!this._outputDesignations.ContainsKey(agent)) - { - this._outputDesignations[agent] = []; - } - } - return (TBuilder)this; - } - - /// - /// Designates the given as sources of intermediate workflow - /// output. See for the defaults-suppression semantics. - /// - public TBuilder WithIntermediateOutputFrom(IEnumerable agents) - { - Throw.IfNull(agents); - this._outputDesignations ??= new(AIAgentIDEqualityComparer.Instance); - foreach (AIAgent agent in agents) - { - Throw.IfNull(agent, nameof(agents)); - if (!this._outputDesignations.TryGetValue(agent, out HashSet? tags)) - { - tags = []; - this._outputDesignations[agent] = tags; - } - tags.Add(OutputTag.Intermediate); - } - return (TBuilder)this; - } - /// /// Adds handoff relationships from a source agent to one or more target agents. /// @@ -683,64 +616,30 @@ public class HandoffWorkflowBuilderCore where TBuilder : HandoffWorkfl }); } - if (!string.IsNullOrWhiteSpace(this._name)) - { - builder.WithName(this._name); - } - - if (!string.IsNullOrWhiteSpace(this._description)) - { - builder.WithDescription(this._description); - } - // Ensure the end executor is bound regardless of whether it ends up as an output // designation source — the user may take full control of output designations. builder.BindExecutor(end); - this.ApplyOutputDesignations(builder, end, executors); - return builder.Build(); - } + // Build the AIAgent -> ExecutorBinding map the base helper expects. + Dictionary agentMap = new(AIAgentIDEqualityComparer.Instance); + foreach (AIAgent agent in this._allAgents) + { + agentMap[agent] = executors[agent.Id]; + } - private void ApplyOutputDesignations( - WorkflowBuilder builder, - HandoffEndExecutor end, - Dictionary executors) - { - if (this._outputDesignations is null) + this.ApplyOutputDesignations(builder, agentMap, "handoff", () => { // Defaults (matches Python's Handoff orchestration): - // end -> terminal output (Output) - // every handoff agent -> intermediate output (Intermediate) + // end -> terminal output + // every handoff agent -> intermediate output builder.WithOutputFrom(end); List agentBindings = [.. executors.Values]; if (agentBindings.Count > 0) { builder.WithIntermediateOutputFrom(agentBindings); } - return; - } + }); - // User took control — replay only their designations. - foreach (AIAgent agent in this._outputDesignations.Keys) - { - if (!executors.TryGetValue(agent.Id, out ExecutorBinding? binding)) - { - throw new InvalidOperationException( - $"Output designation references agent '{agent.Name ?? agent.Id}', which is not a participant in this handoff workflow."); - } - - HashSet tags = this._outputDesignations[agent]; - if (tags.Count == 0) - { - builder.WithOutputFrom(binding); - } - else - { - foreach (OutputTag tag in tags) - { - builder.WithOutputFrom(binding, tag); - } - } - } + return builder.Build(); } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/MagenticWorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/MagenticWorkflowBuilder.cs index 0c8782ceb2..3f3cf005f5 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/MagenticWorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/MagenticWorkflowBuilder.cs @@ -29,18 +29,14 @@ namespace Microsoft.Agents.AI.Workflows; /// /// [Experimental(DiagnosticConstants.ExperimentalFeatureDiagnostic)] -public class MagenticWorkflowBuilder(AIAgent managerAgent) +public class MagenticWorkflowBuilder(AIAgent managerAgent) : OrchestrationBuilderBase { private readonly List _team = new(); - private string? _name; - private string? _description; private int _maxStalls = TaskLimits.DefaultMaxStallCount; private int? _maxRounds; private int? _maxResets; private bool _requirePlanSignoff = true; - private Dictionary>? _outputDesignations; - /// public MagenticWorkflowBuilder AddParticipants(params IEnumerable agents) { @@ -48,20 +44,6 @@ public class MagenticWorkflowBuilder(AIAgent managerAgent) return this; } - /// - public MagenticWorkflowBuilder WithName(string name) - { - this._name = name; - return this; - } - - /// - public MagenticWorkflowBuilder WithDescription(string description) - { - this._description = description; - return this; - } - /// /// Set the maximum number of coordination rounds. means unlimited. /// @@ -103,48 +85,6 @@ public class MagenticWorkflowBuilder(AIAgent managerAgent) return this; } - /// - /// Designates the given as sources of terminal workflow output. - /// Calling any output-designation method (this or ) - /// suppresses the orchestration-specific defaults: only the user-specified designations - /// reach the inner . - /// - public MagenticWorkflowBuilder WithOutputFrom(params IEnumerable agents) - { - Throw.IfNull(agents); - this._outputDesignations ??= new(AIAgentIDEqualityComparer.Instance); - foreach (AIAgent agent in agents) - { - Throw.IfNull(agent, nameof(agents)); - if (!this._outputDesignations.ContainsKey(agent)) - { - this._outputDesignations[agent] = []; - } - } - return this; - } - - /// - /// Designates the given as sources of intermediate workflow output. - /// See for the defaults-suppression semantics. - /// - public MagenticWorkflowBuilder WithIntermediateOutputFrom(IEnumerable agents) - { - Throw.IfNull(agents); - this._outputDesignations ??= new(AIAgentIDEqualityComparer.Instance); - foreach (AIAgent agent in agents) - { - Throw.IfNull(agent, nameof(agents)); - if (!this._outputDesignations.TryGetValue(agent, out HashSet? tags)) - { - tags = []; - this._outputDesignations[agent] = tags; - } - tags.Add(OutputTag.Intermediate); - } - return this; - } - private WorkflowBuilder ReduceToWorkflowBuilder() { // Create a copy of the team so that improper modifications by using the builder after .Build() do not affect the @@ -172,60 +112,18 @@ public class MagenticWorkflowBuilder(AIAgent managerAgent) } result.AddFanOutEdge(orchestrator, teamBindings); - this.ApplyOutputDesignations(result, orchestrator, teamMap); - if (!string.IsNullOrWhiteSpace(this._name)) + this.ApplyOutputDesignations(result, teamMap, "Magentic", () => { - result.WithName(this._name); - } - - if (!string.IsNullOrWhiteSpace(this._description)) - { - result.WithDescription(this._description); - } - - return result; - } - - private void ApplyOutputDesignations( - WorkflowBuilder builder, - ExecutorBinding orchestrator, - Dictionary teamMap) - { - if (this._outputDesignations is null) - { - // Defaults (matches Python Magentic orchestration): - // orchestrator -> terminal output - // team members -> intermediate output - builder.WithOutputFrom(orchestrator); + result.WithOutputFrom(orchestrator); if (teamMap.Count > 0) { - builder.WithIntermediateOutputFrom([.. teamMap.Values]); + result.WithIntermediateOutputFrom([.. teamMap.Values]); } - return; - } + }); - foreach (AIAgent agent in this._outputDesignations.Keys) - { - if (!teamMap.TryGetValue(agent, out ExecutorBinding? binding)) - { - throw new InvalidOperationException( - $"Output designation references agent '{agent.Name ?? agent.Id}', which is not a participant in this Magentic workflow."); - } - - HashSet tags = this._outputDesignations[agent]; - if (tags.Count == 0) - { - builder.WithOutputFrom(binding); - } - else - { - foreach (OutputTag tag in tags) - { - builder.WithOutputFrom(binding, tag); - } - } - } + this.ApplyMetadata(result); + return result; } /// diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/OrchestrationBuilderBase.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/OrchestrationBuilderBase.cs new file mode 100644 index 0000000000..2e0464d405 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/OrchestrationBuilderBase.cs @@ -0,0 +1,154 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Agents.AI.Workflows; + +/// +/// Common fluent surface shared by every orchestration-style workflow builder: +/// human-readable name + description, and the +/// / output-designation +/// pair with memoized defaults-suppression semantics. +/// +/// The concrete builder type, for fluent self-return. +public abstract class OrchestrationBuilderBase + where TBuilder : OrchestrationBuilderBase +{ + /// Optional workflow name; applied to the inner at Build(). + protected string? Name { get; private set; } + + /// Optional workflow description; applied to the inner at Build(). + protected string? Description { get; private set; } + + /// + /// Memoized output designations. means the user has not made any + /// explicit designation, and the orchestration-specific defaults will be applied at + /// Build() time. A non- (possibly empty) map means the user took + /// control and only these designations will be replayed onto the inner + /// . An entry's value is the set of tags requested for the + /// agent — an empty set encodes a terminal-only designation. + /// + protected Dictionary>? OutputDesignations { get; private set; } + + /// Sets the human-readable name for the workflow. + public TBuilder WithName(string name) + { + this.Name = name; + return (TBuilder)this; + } + + /// Sets the description for the workflow. + public TBuilder WithDescription(string description) + { + this.Description = description; + return (TBuilder)this; + } + + /// + /// Designates the given as sources of terminal workflow output. + /// Calling any output-designation method (this or ) + /// suppresses the orchestration-specific defaults: only the user-specified designations + /// reach the inner . + /// + public TBuilder WithOutputFrom(params IEnumerable agents) + { + Throw.IfNull(agents); + this.OutputDesignations ??= new(AIAgentIDEqualityComparer.Instance); + foreach (AIAgent agent in agents) + { + Throw.IfNull(agent, nameof(agents)); + if (!this.OutputDesignations.ContainsKey(agent)) + { + this.OutputDesignations[agent] = []; + } + } + return (TBuilder)this; + } + + /// + /// Designates the given as sources of intermediate workflow + /// output. See for the defaults-suppression semantics. + /// + public TBuilder WithIntermediateOutputFrom(IEnumerable agents) + { + Throw.IfNull(agents); + this.OutputDesignations ??= new(AIAgentIDEqualityComparer.Instance); + foreach (AIAgent agent in agents) + { + Throw.IfNull(agent, nameof(agents)); + if (!this.OutputDesignations.TryGetValue(agent, out HashSet? tags)) + { + tags = []; + this.OutputDesignations[agent] = tags; + } + tags.Add(OutputTag.Intermediate); + } + return (TBuilder)this; + } + + /// + /// Applies the optional and to . + /// Subclasses should call this from their Build() implementation. + /// + protected void ApplyMetadata(WorkflowBuilder builder) + { + Throw.IfNull(builder); + if (!string.IsNullOrWhiteSpace(this.Name)) + { + builder.WithName(this.Name!); + } + if (!string.IsNullOrWhiteSpace(this.Description)) + { + builder.WithDescription(this.Description!); + } + } + + /// + /// Applies the user's memoized output designations to , or invokes + /// if the user made no explicit designation. + /// + /// The inner . + /// Map from participating to its bound executor. + /// Used in the not-a-participant error message (e.g. "sequential", "group chat"). + /// Action invoked when no explicit designation was made. + protected void ApplyOutputDesignations( + WorkflowBuilder builder, + IReadOnlyDictionary agentMap, + string orchestrationKind, + Action applyDefaults) + { + Throw.IfNull(builder); + Throw.IfNull(agentMap); + Throw.IfNull(applyDefaults); + + if (this.OutputDesignations is null) + { + applyDefaults(); + return; + } + + foreach (AIAgent agent in this.OutputDesignations.Keys) + { + if (!agentMap.TryGetValue(agent, out ExecutorBinding? binding)) + { + throw new InvalidOperationException( + $"Output designation references agent '{agent.Name ?? agent.Id}', which is not a participant in this {orchestrationKind} workflow."); + } + + HashSet tags = this.OutputDesignations[agent]; + if (tags.Count == 0) + { + builder.WithOutputFrom(binding); + } + else + { + foreach (OutputTag tag in tags) + { + builder.WithOutputFrom(binding, tag); + } + } + } + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/SequentialWorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/SequentialWorkflowBuilder.cs new file mode 100644 index 0000000000..18019032d1 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/SequentialWorkflowBuilder.cs @@ -0,0 +1,85 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Linq; +using Microsoft.Agents.AI.Workflows.Specialized; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Agents.AI.Workflows; + +/// +/// Fluent builder for sequential agent workflows: a pipeline where the output of one +/// agent is the input to the next, terminating in an aggregator that yields the +/// accumulated s as the workflow output. +/// +/// +/// When no explicit output designations are made, the default is the Python-aligned +/// shape: the terminal aggregator is the workflow output, and every participating agent +/// is designated as an intermediate output source. Calling +/// +/// or +/// at all suppresses these defaults. +/// +public sealed class SequentialWorkflowBuilder : OrchestrationBuilderBase +{ + private readonly List _agents = []; + + /// + /// Initializes a new with the given pipeline + /// of . + /// + public SequentialWorkflowBuilder(params IEnumerable agents) + { + Throw.IfNull(agents); + foreach (AIAgent agent in agents) + { + Throw.IfNull(agent, nameof(agents)); + this._agents.Add(agent); + } + } + + /// Builds the configured sequential workflow. + public Workflow Build() + { + if (this._agents.Count == 0) + { + throw new ArgumentException("At least one agent must be provided to the SequentialWorkflowBuilder.", "agents"); + } + + AIAgentHostOptions options = new() + { + ReassignOtherAgentsAsUsers = true, + ForwardIncomingMessages = true, + }; + + Dictionary agentMap = new(AIAgentIDEqualityComparer.Instance); + List agentExecutors = new(this._agents.Count); + foreach (AIAgent agent in this._agents) + { + ExecutorBinding binding = agent.BindAsExecutor(options); + agentExecutors.Add(binding); + agentMap[agent] = binding; + } + + ExecutorBinding previous = agentExecutors[0]; + WorkflowBuilder builder = new(previous); + foreach (ExecutorBinding next in agentExecutors.Skip(1)) + { + builder.AddEdge(previous, next); + previous = next; + } + + OutputMessagesExecutor end = new(); + builder.AddEdge(previous, end).BindExecutor(end); + + this.ApplyMetadata(builder); + this.ApplyOutputDesignations(builder, agentMap, "sequential", () => + { + builder.WithOutputFrom(end); + builder.WithIntermediateOutputFrom(agentExecutors); + }); + + return builder.Build(); + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilder.ConcurrentTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilder.ConcurrentTests.cs deleted file mode 100644 index eeb078cb0c..0000000000 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilder.ConcurrentTests.cs +++ /dev/null @@ -1,77 +0,0 @@ -// Copyright (c) Microsoft. All rights reserved. - -using System; -using System.Collections.Generic; -using System.Linq; -using System.Runtime.CompilerServices; -using System.Text.RegularExpressions; -using System.Threading.Tasks; -using FluentAssertions; -using Microsoft.Extensions.AI; - -#pragma warning disable SYSLIB1045 // Use GeneratedRegex -#pragma warning disable RCS1186 // Use Regex instance instead of static method - -namespace Microsoft.Agents.AI.Workflows.UnitTests; - -public static partial class AgentWorkflowBuilderTests -{ - public class ConcurrentTests - { - [Fact] - public void BuildConcurrent_InvalidArguments_Throws() - { - Assert.Throws("agents", () => AgentWorkflowBuilder.BuildConcurrent(null!)); - } - - [Fact] - public async Task BuildConcurrent_AgentsRunInParallelAsync() - { - StrongBox> barrier = new(); - StrongBox remaining = new(); - - var workflow = AgentWorkflowBuilder.BuildConcurrent( - [ - new DoubleEchoAgentWithBarrier("agent1", barrier, remaining), - new DoubleEchoAgentWithBarrier("agent2", barrier, remaining), - ]); - - for (int iter = 0; iter < 3; iter++) - { - barrier.Value = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - remaining.Value = 2; - - (string updateText, List? result, _, _) = await RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, "abc")]); - Assert.NotEmpty(updateText); - Assert.NotNull(result); - - // TODO: https://github.com/microsoft/agent-framework/issues/784 - // These asserts are flaky until we guarantee message delivery order. - Assert.Single(Regex.Matches(updateText, "agent1")); - Assert.Single(Regex.Matches(updateText, "agent2")); - Assert.Equal(4, Regex.Matches(updateText, "abc").Count); - Assert.Equal(2, result.Count); - } - } - - [Fact] - public void Test_BuildConcurrent_DefaultDesignationsMatchSpec() - { - Workflow workflow = AgentWorkflowBuilder.BuildConcurrent( - [new DoubleEchoAgent("agent1"), new DoubleEchoAgent("agent2"), new DoubleEchoAgent("agent3")]); - - Dictionary> designations = workflow.OutputExecutors; - - List>> terminals = designations - .Where(kvp => kvp.Value.Count == 0) - .ToList(); - terminals.Should().ContainSingle("Concurrent has exactly one terminal output executor (ConcurrentEndExecutor)"); - - List>> intermediates = designations - .Where(kvp => kvp.Value.Contains(OutputTag.Intermediate)) - .ToList(); - intermediates.Should().HaveCount(6, - "every concurrent agent (3) and its per-agent accumulator (3) are designated intermediate"); - } - } -} diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilder.SequentialTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilder.SequentialTests.cs deleted file mode 100644 index 773acfefdf..0000000000 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilder.SequentialTests.cs +++ /dev/null @@ -1,90 +0,0 @@ -// Copyright (c) Microsoft. All rights reserved. - -using System; -using System.Collections.Generic; -using System.Linq; -using System.Threading.Tasks; -using FluentAssertions; -using Microsoft.Extensions.AI; - -namespace Microsoft.Agents.AI.Workflows.UnitTests; - -public static partial class AgentWorkflowBuilderTests -{ - public class SequentialTests - { - [Fact] - public void BuildSequential_InvalidArguments_Throws() - { - Assert.Throws("agents", () => AgentWorkflowBuilder.BuildSequential(workflowName: null!, null!)); - Assert.Throws("agents", () => AgentWorkflowBuilder.BuildSequential()); - } - - [Theory] - [InlineData(1)] - [InlineData(2)] - [InlineData(3)] - [InlineData(4)] - [InlineData(5)] - public async Task BuildSequential_AgentsRunInOrderAsync(int numAgents) - { - var workflow = AgentWorkflowBuilder.BuildSequential( - from i in Enumerable.Range(1, numAgents) - select new DoubleEchoAgent($"agent{i}")); - - for (int iter = 0; iter < 3; iter++) - { - const string UserInput = "abc"; - (string updateText, List? result, _, _) = await RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, UserInput)]); - - Assert.NotNull(result); - Assert.Equal(numAgents + 1, result.Count); - - Assert.Equal(ChatRole.User, result[0].Role); - Assert.Null(result[0].AuthorName); - Assert.Equal(UserInput, result[0].Text); - - string[] texts = new string[numAgents + 1]; - texts[0] = UserInput; - string expectedTotal = string.Empty; - for (int i = 1; i < numAgents + 1; i++) - { - string id = $"agent{((i - 1) % numAgents) + 1}"; - texts[i] = $"{id}{Double(string.Concat(texts.Take(i)))}"; - Assert.Equal(ChatRole.Assistant, result[i].Role); - Assert.Equal(id, result[i].AuthorName); - Assert.Equal(texts[i], result[i].Text); - expectedTotal += texts[i]; - } - - Assert.Equal(expectedTotal, updateText); - Assert.Equal(UserInput + expectedTotal, string.Concat(result)); - - static string Double(string s) => s + s; - } - } - - [Fact] - public void Test_BuildSequential_DefaultDesignationsMatchSpec() - { - Workflow workflow = AgentWorkflowBuilder.BuildSequential( - new DoubleEchoAgent("agent1"), - new DoubleEchoAgent("agent2"), - new DoubleEchoAgent("agent3")); - - // Defaults: every agent executor is intermediate; exactly one terminal entry (the OutputMessagesExecutor). - Dictionary> designations = workflow.OutputExecutors; - designations.Should().NotBeEmpty(); - - List>> terminals = designations - .Where(kvp => kvp.Value.Count == 0) - .ToList(); - terminals.Should().ContainSingle("Sequential has exactly one terminal output executor (OutputMessagesExecutor)"); - - List>> intermediates = designations - .Where(kvp => kvp.Value.Contains(OutputTag.Intermediate)) - .ToList(); - intermediates.Should().HaveCount(3, "every agent in the pipeline is designated intermediate"); - } - } -} diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilderTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilderTests.cs index bb17c8644d..d615b8bc74 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilderTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilderTests.cs @@ -4,132 +4,174 @@ using System; using System.Collections.Generic; using System.Linq; using System.Runtime.CompilerServices; -using System.Text; -using System.Text.Json; -using System.Threading; +using System.Text.RegularExpressions; using System.Threading.Tasks; -using Microsoft.Agents.AI.Workflows.InProc; +using FluentAssertions; using Microsoft.Extensions.AI; +#pragma warning disable SYSLIB1045 // Use GeneratedRegex +#pragma warning disable RCS1186 // Use Regex instance instead of static method + namespace Microsoft.Agents.AI.Workflows.UnitTests; /// -/// Container for tests covering entry points that -/// do not produce a dedicated builder type (currently BuildSequential and -/// BuildConcurrent). The actual test methods live in nested classes -/// ( and ) split across -/// partial files. Shared test helpers — the DoubleEchoAgent family and the -/// RunWorkflow* methods — are declared on this outer partial as -/// internal so the nested test classes and the standalone -/// can all reuse them. +/// Tests targeting the static helper surface — +/// , +/// , +/// and the various Create*BuilderWith factories. Per-builder unit tests live in their own +/// files (, , etc.). /// -public static partial class AgentWorkflowBuilderTests +public class AgentWorkflowBuilderTests { - internal class DoubleEchoAgent(string name) : AIAgent + [Fact] + public void Test_AgentWorkflowBuilder_BuildSequential_InvalidArguments_Throws() { - public override string Name => name; - - protected override ValueTask CreateSessionCoreAsync(CancellationToken cancellationToken = default) - => new(new DoubleEchoAgentSession()); - - protected override ValueTask DeserializeSessionCoreAsync(JsonElement serializedState, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default) - => new(new DoubleEchoAgentSession()); - - protected override ValueTask SerializeSessionCoreAsync(AgentSession session, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default) - => default; - - protected override Task RunCoreAsync( - IEnumerable messages, AgentSession? session = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default) => - throw new NotImplementedException(); - - protected override async IAsyncEnumerable RunCoreStreamingAsync( - IEnumerable messages, AgentSession? session = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) - { - await Task.Yield(); - - var contents = messages.SelectMany(m => m.Contents).ToList(); - string id = Guid.NewGuid().ToString("N"); - yield return new AgentResponseUpdate(ChatRole.Assistant, this.Name) { AuthorName = this.Name, MessageId = id }; - yield return new AgentResponseUpdate(ChatRole.Assistant, contents) { AuthorName = this.Name, MessageId = id }; - yield return new AgentResponseUpdate(ChatRole.Assistant, contents) { AuthorName = this.Name, MessageId = id }; - } + Assert.Throws("agents", () => AgentWorkflowBuilder.BuildSequential(workflowName: null!, null!)); + Assert.Throws("agents", () => AgentWorkflowBuilder.BuildSequential()); } - internal sealed class DoubleEchoAgentSession() : AgentSession(); - - internal sealed class DoubleEchoAgentWithBarrier(string name, StrongBox> barrier, StrongBox remaining) : DoubleEchoAgent(name) + [Theory] + [InlineData(1)] + [InlineData(2)] + [InlineData(3)] + public async Task Test_AgentWorkflowBuilder_BuildSequential_DelegatesToBuilderAsync(int numAgents) { - protected override async IAsyncEnumerable RunCoreStreamingAsync( - IEnumerable messages, AgentSession? session = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) - { - if (Interlocked.Decrement(ref remaining.Value) == 0) - { - barrier.Value!.SetResult(true); - } + Workflow workflow = AgentWorkflowBuilder.BuildSequential( + from i in Enumerable.Range(1, numAgents) + select new OrchestrationTestHelpers.DoubleEchoAgent($"agent{i}")); - await barrier.Value!.Task.ConfigureAwait(false); + // Smoke: end-to-end run produces a non-empty result. Detailed pipeline-ordering + // assertions live in SequentialWorkflowBuilderTests. + (string updateText, List? result, _, _) = + await OrchestrationTestHelpers.RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, "abc")]); - await foreach (var update in base.RunCoreStreamingAsync(messages, session, options, cancellationToken)) - { - await Task.Yield(); - yield return update; - } - } + Assert.NotNull(result); + Assert.Equal(numAgents + 1, result.Count); + Assert.NotEmpty(updateText); } - internal sealed record WorkflowRunResult(string UpdateText, List? Result, CheckpointInfo? LastCheckpoint, List PendingRequests); - - internal static async Task RunWorkflowCheckpointedAsync( - Workflow workflow, List input, InProcessExecutionEnvironment environment, CheckpointInfo? fromCheckpoint = null) + [Fact] + public void Test_AgentWorkflowBuilder_BuildSequential_WithWorkflowNameSetsNameOnWorkflow() { - await using StreamingRun run = - fromCheckpoint != null ? await environment.ResumeStreamingAsync(workflow, fromCheckpoint) - : await environment.OpenStreamingAsync(workflow); + Workflow workflow = AgentWorkflowBuilder.BuildSequential( + "static-sequential", + new OrchestrationTestHelpers.DoubleEchoAgent("agent1")); - await run.TrySendMessageAsync(input); - await run.TrySendMessageAsync(new TurnToken(emitEvents: true)); - - return await ProcessWorkflowRunAsync(run); + workflow.Name.Should().Be("static-sequential"); } - internal static async Task ProcessWorkflowRunAsync(StreamingRun run) + [Fact] + public void Test_AgentWorkflowBuilder_BuildConcurrent_InvalidArguments_Throws() { - StringBuilder sb = new(); - WorkflowOutputEvent? output = null; - CheckpointInfo? lastCheckpoint = null; - - List pendingRequests = []; - - await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false).ConfigureAwait(false)) - { - switch (evt) - { - case AgentResponseUpdateEvent responseUpdate: - sb.Append(responseUpdate.Data); - break; - - case RequestInfoEvent requestInfo: - pendingRequests.Add(requestInfo); - break; - - case WorkflowOutputEvent e: - output = e; - break; - - case WorkflowErrorEvent errorEvent: - Assert.Fail($"Workflow execution failed with error: {errorEvent.Exception}"); - break; - - case SuperStepCompletedEvent stepCompleted: - lastCheckpoint = stepCompleted.CompletionInfo?.Checkpoint; - break; - } - } - - return new(sb.ToString(), output?.As>(), lastCheckpoint, pendingRequests); + Assert.Throws("agents", () => AgentWorkflowBuilder.BuildConcurrent(null!)); } - internal static Task RunWorkflowAsync( - Workflow workflow, List input, ExecutionEnvironment executionEnvironment = ExecutionEnvironment.InProcess_Lockstep) - => RunWorkflowCheckpointedAsync(workflow, input, executionEnvironment.ToWorkflowExecutionEnvironment()); -} \ No newline at end of file + [Fact] + public async Task Test_AgentWorkflowBuilder_BuildConcurrent_DelegatesToBuilderAsync() + { + StrongBox> barrier = new(); + StrongBox remaining = new(); + + Workflow workflow = AgentWorkflowBuilder.BuildConcurrent( + [ + new OrchestrationTestHelpers.DoubleEchoAgentWithBarrier("agent1", barrier, remaining), + new OrchestrationTestHelpers.DoubleEchoAgentWithBarrier("agent2", barrier, remaining), + ]); + + barrier.Value = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + remaining.Value = 2; + + (string updateText, List? result, _, _) = + await OrchestrationTestHelpers.RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, "abc")]); + + Assert.NotEmpty(updateText); + Assert.NotNull(result); + Assert.Equal(2, result.Count); + Assert.Single(Regex.Matches(updateText, "agent1")); + Assert.Single(Regex.Matches(updateText, "agent2")); + } + + [Fact] + public void Test_AgentWorkflowBuilder_BuildConcurrent_WithWorkflowNameSetsNameOnWorkflow() + { + Workflow workflow = AgentWorkflowBuilder.BuildConcurrent( + "static-concurrent", + [new OrchestrationTestHelpers.DoubleEchoAgent("agent1")]); + + workflow.Name.Should().Be("static-concurrent"); + } + + [Fact] + public async Task Test_AgentWorkflowBuilder_BuildConcurrent_AggregatorIsHonoredAsync() + { + // Replace the default ("last message from each agent") with a custom aggregator, + // and confirm the workflow yields its result. + List sentinel = [new(ChatRole.Assistant, "custom-aggregator-result")]; + + Workflow workflow = AgentWorkflowBuilder.BuildConcurrent( + [new OrchestrationTestHelpers.DoubleEchoAgent("agent1")], + aggregator: _ => sentinel); + + (_, List? result, _, _) = + await OrchestrationTestHelpers.RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, "abc")]); + + result.Should().NotBeNull().And.ContainSingle(); + result![0].Text.Should().Be("custom-aggregator-result"); + } + + [Fact] + public void Test_AgentWorkflowBuilder_CreateSequentialBuilderWith_RejectsNull() + { + Assert.Throws("agents", () => AgentWorkflowBuilder.CreateSequentialBuilderWith(null!)); + } + + [Fact] + public void Test_AgentWorkflowBuilder_CreateSequentialBuilderWith_ReturnsConfigurableBuilder() + { + OrchestrationTestHelpers.DoubleEchoAgent agent = new("agent1"); + + SequentialWorkflowBuilder builder = AgentWorkflowBuilder.CreateSequentialBuilderWith(agent); + Workflow workflow = builder.WithName("via-factory").Build(); + + workflow.Name.Should().Be("via-factory"); + } + + [Fact] + public void Test_AgentWorkflowBuilder_CreateConcurrentBuilderWith_RejectsNull() + { + Assert.Throws("agents", () => AgentWorkflowBuilder.CreateConcurrentBuilderWith(null!)); + } + + [Fact] + public void Test_AgentWorkflowBuilder_CreateConcurrentBuilderWith_ReturnsConfigurableBuilder() + { + OrchestrationTestHelpers.DoubleEchoAgent agent = new("agent1"); + + ConcurrentWorkflowBuilder builder = AgentWorkflowBuilder.CreateConcurrentBuilderWith(agent); + Workflow workflow = builder.WithName("via-factory").Build(); + + workflow.Name.Should().Be("via-factory"); + } + + [Fact] + public void Test_AgentWorkflowBuilder_CreateHandoffBuilderWith_RejectsNull() + { +#pragma warning disable MAAIW001 + Assert.Throws("initialAgent", () => AgentWorkflowBuilder.CreateHandoffBuilderWith(null!)); +#pragma warning restore MAAIW001 + } + + [Fact] + public void Test_AgentWorkflowBuilder_CreateGroupChatBuilderWith_RejectsNull() + { + Assert.Throws("managerFactory", () => AgentWorkflowBuilder.CreateGroupChatBuilderWith(null!)); + } + + [Fact] + public void Test_AgentWorkflowBuilder_CreateMagenticBuilderWith_RejectsNull() + { +#pragma warning disable MAAIW001 + Assert.Throws("managerAgent", () => AgentWorkflowBuilder.CreateMagenticBuilderWith(null!)); +#pragma warning restore MAAIW001 + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ConcurrentWorkflowBuilderTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ConcurrentWorkflowBuilderTests.cs new file mode 100644 index 0000000000..c03a98bcec --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ConcurrentWorkflowBuilderTests.cs @@ -0,0 +1,165 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Runtime.CompilerServices; +using System.Text.RegularExpressions; +using System.Threading.Tasks; +using FluentAssertions; +using Microsoft.Agents.AI.Workflows.UnitTests.Futures; +using Microsoft.Extensions.AI; +using Xunit; + +#pragma warning disable SYSLIB1045 // Use GeneratedRegex +#pragma warning disable RCS1186 // Use Regex instance instead of static method + +namespace Microsoft.Agents.AI.Workflows.UnitTests; + +public class ConcurrentWorkflowBuilderTests +{ + [Fact] + public void Test_ConcurrentWorkflowBuilder_InvalidArguments_Throws() + { + Assert.Throws("agents", () => new ConcurrentWorkflowBuilder(null!)); + Assert.Throws("agents", () => new ConcurrentWorkflowBuilder().Build()); + + Assert.Throws("agents", () => AgentWorkflowBuilder.BuildConcurrent(null!)); + Assert.Throws("agents", () => AgentWorkflowBuilder.CreateConcurrentBuilderWith(null!)); + } + + [Fact] + public async Task Test_ConcurrentWorkflowBuilder_AgentsRunInParallelAsync() + { + StrongBox> barrier = new(); + StrongBox remaining = new(); + + var workflow = new ConcurrentWorkflowBuilder( + new OrchestrationTestHelpers.DoubleEchoAgentWithBarrier("agent1", barrier, remaining), + new OrchestrationTestHelpers.DoubleEchoAgentWithBarrier("agent2", barrier, remaining)) + .Build(); + + for (int iter = 0; iter < 3; iter++) + { + barrier.Value = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + remaining.Value = 2; + + (string updateText, List? result, _, _) = + await OrchestrationTestHelpers.RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, "abc")]); + Assert.NotEmpty(updateText); + Assert.NotNull(result); + + // TODO: https://github.com/microsoft/agent-framework/issues/784 + // These asserts are flaky until we guarantee message delivery order. + Assert.Single(Regex.Matches(updateText, "agent1")); + Assert.Single(Regex.Matches(updateText, "agent2")); + Assert.Equal(4, Regex.Matches(updateText, "abc").Count); + Assert.Equal(2, result.Count); + } + } + + [Fact] + public void Test_ConcurrentWorkflowBuilder_DefaultDesignationsMatchSpec() + { + Workflow workflow = new ConcurrentWorkflowBuilder( + new OrchestrationTestHelpers.DoubleEchoAgent("agent1"), + new OrchestrationTestHelpers.DoubleEchoAgent("agent2"), + new OrchestrationTestHelpers.DoubleEchoAgent("agent3")) + .Build(); + + Dictionary> designations = workflow.OutputExecutors; + designations.Where(kvp => kvp.Value.Count == 0) + .Should().ContainSingle("ConcurrentEndExecutor is the sole terminal output by default"); + designations.Where(kvp => kvp.Value.Contains(OutputTag.Intermediate)) + .Should().HaveCount(6, "every agent (3) and per-agent accumulator (3) is designated intermediate by default"); + } + + [Fact] + public void Test_ConcurrentWorkflowBuilder_ExplicitDesignationsReplaceDefaults() + { + OrchestrationTestHelpers.DoubleEchoAgent a1 = new("agent1"); + OrchestrationTestHelpers.DoubleEchoAgent a2 = new("agent2"); + OrchestrationTestHelpers.DoubleEchoAgent a3 = new("agent3"); + + Workflow workflow = new ConcurrentWorkflowBuilder(a1, a2, a3) + .WithOutputFrom(a1) + .WithIntermediateOutputFrom([a2]) + .Build(); + + Dictionary> designations = workflow.OutputExecutors; + + designations.Should().HaveCount(2, + "only the two explicitly-designated agents land on the inner builder; the end + accumulator defaults are suppressed"); + designations.Values.Where(tags => tags.Count == 0) + .Should().ContainSingle("agent1 is the only terminal designation"); + designations.Values.Where(tags => tags.Contains(OutputTag.Intermediate)) + .Should().ContainSingle("agent2 is the only intermediate designation"); + } + + [Fact] + public void Test_ConcurrentWorkflowBuilder_DesignationForNonParticipantThrows() + { + OrchestrationTestHelpers.DoubleEchoAgent participant = new("p1"); + OrchestrationTestHelpers.DoubleEchoAgent stranger = new("stranger"); + + ConcurrentWorkflowBuilder builder = new ConcurrentWorkflowBuilder(participant) + .WithIntermediateOutputFrom([stranger]); + + Action build = () => builder.Build(); + build.Should().Throw().WithMessage("*stranger*"); + } + + [Fact] + public void Test_ConcurrentWorkflowBuilder_WithNamePropagatesToWorkflow() + { + Workflow workflow = new ConcurrentWorkflowBuilder(new OrchestrationTestHelpers.DoubleEchoAgent("agent1")) + .WithName("named-concurrent") + .Build(); + + workflow.Name.Should().Be("named-concurrent"); + } + + [Fact] + public void Test_ConcurrentWorkflowBuilder_WithDescriptionPropagatesToWorkflow() + { + Workflow workflow = new ConcurrentWorkflowBuilder(new OrchestrationTestHelpers.DoubleEchoAgent("agent1")) + .WithDescription("describes the concurrent fan-out/fan-in") + .Build(); + + workflow.Description.Should().Be("describes the concurrent fan-out/fan-in"); + } + + [Collection(FuturesSerialCollection.Name)] + public class AsAgentForwarding + { + [Fact] + public async Task Test_ConcurrentWorkflowBuilder_AsAgent_OnlyTerminalDesignationSurfacesAsync() + { + using FuturesScope _ = new(enabled: true); + + OrchestrationTestHelpers.DoubleEchoAgent agent1 = new("agent1"); + OrchestrationTestHelpers.DoubleEchoAgent agent2 = new("agent2"); + + // Designate only agent1 as a terminal output source — agent2 and the fan-in + // aggregator default-intermediate designations are suppressed. + Workflow workflow = new ConcurrentWorkflowBuilder(agent1, agent2) + .WithOutputFrom(agent1) + .Build(); + + List updates = await workflow + .AsAIAgent("WorkflowAgent") + .RunStreamingAsync(new ChatMessage(ChatRole.User, "abc")) + .ToListAsync(); + + HashSet authoredBy = updates + .Select(u => u.AuthorName) + .Where(n => !string.IsNullOrEmpty(n)) + .Select(n => n!) + .ToHashSet(); + + authoredBy.Should().Contain("agent1", "the designated agent must surface"); + authoredBy.Should().NotContain("agent2", + "the undesignated agent must not surface when only one is designated under Futures-on"); + } + } +} diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/GroupChatWorkflowBuilderTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/GroupChatWorkflowBuilderTests.cs index 1112bbe552..02cd9907a5 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/GroupChatWorkflowBuilderTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/GroupChatWorkflowBuilderTests.cs @@ -20,11 +20,11 @@ public class GroupChatWorkflowBuilderTests { Assert.Throws("managerFactory", () => AgentWorkflowBuilder.CreateGroupChatBuilderWith(null!)); - var groupChat = AgentWorkflowBuilder.CreateGroupChatBuilderWith(_ => new RoundRobinGroupChatManager([new AgentWorkflowBuilderTests.DoubleEchoAgent("a1")])); + var groupChat = AgentWorkflowBuilder.CreateGroupChatBuilderWith(_ => new RoundRobinGroupChatManager([new OrchestrationTestHelpers.DoubleEchoAgent("a1")])); Assert.NotNull(groupChat); Assert.Throws("agents", () => groupChat.AddParticipants(null!)); Assert.Throws("agents", () => groupChat.AddParticipants([null!])); - Assert.Throws("agents", () => groupChat.AddParticipants(new AgentWorkflowBuilderTests.DoubleEchoAgent("a1"), null!)); + Assert.Throws("agents", () => groupChat.AddParticipants(new OrchestrationTestHelpers.DoubleEchoAgent("a1"), null!)); Assert.Throws("agents", () => new RoundRobinGroupChatManager(null!)); } @@ -32,7 +32,7 @@ public class GroupChatWorkflowBuilderTests [Fact] public void GroupChatManager_MaximumIterationCount_Invalid_Throws() { - var manager = new RoundRobinGroupChatManager([new AgentWorkflowBuilderTests.DoubleEchoAgent("a1")]); + var manager = new RoundRobinGroupChatManager([new OrchestrationTestHelpers.DoubleEchoAgent("a1")]); const int DefaultMaxIterations = 40; Assert.Equal(DefaultMaxIterations, manager.MaximumIterationCount); @@ -58,7 +58,7 @@ public class GroupChatWorkflowBuilderTests var workflow = AgentWorkflowBuilder .CreateGroupChatBuilderWith(agents => new RoundRobinGroupChatManager(agents) { MaximumIterationCount = 2 }) - .AddParticipants(new AgentWorkflowBuilderTests.DoubleEchoAgent("agent1"), new AgentWorkflowBuilderTests.DoubleEchoAgent("agent2")) + .AddParticipants(new OrchestrationTestHelpers.DoubleEchoAgent("agent1"), new OrchestrationTestHelpers.DoubleEchoAgent("agent2")) .WithName(WorkflowName) .WithDescription(WorkflowDescription) .Build(); @@ -74,7 +74,7 @@ public class GroupChatWorkflowBuilderTests var workflow = AgentWorkflowBuilder .CreateGroupChatBuilderWith(agents => new RoundRobinGroupChatManager(agents) { MaximumIterationCount = 2 }) - .AddParticipants(new AgentWorkflowBuilderTests.DoubleEchoAgent("agent1")) + .AddParticipants(new OrchestrationTestHelpers.DoubleEchoAgent("agent1")) .WithName(WorkflowName) .Build(); @@ -87,7 +87,7 @@ public class GroupChatWorkflowBuilderTests { var workflow = AgentWorkflowBuilder .CreateGroupChatBuilderWith(agents => new RoundRobinGroupChatManager(agents) { MaximumIterationCount = 2 }) - .AddParticipants(new AgentWorkflowBuilderTests.DoubleEchoAgent("agent1")) + .AddParticipants(new OrchestrationTestHelpers.DoubleEchoAgent("agent1")) .Build(); Assert.Null(workflow.Name); @@ -104,14 +104,14 @@ public class GroupChatWorkflowBuilderTests { const int NumAgents = 3; var workflow = AgentWorkflowBuilder.CreateGroupChatBuilderWith(agents => new RoundRobinGroupChatManager(agents) { MaximumIterationCount = maxIterations }) - .AddParticipants(new AgentWorkflowBuilderTests.DoubleEchoAgent("agent1"), new AgentWorkflowBuilderTests.DoubleEchoAgent("agent2")) - .AddParticipants(new AgentWorkflowBuilderTests.DoubleEchoAgent("agent3")) + .AddParticipants(new OrchestrationTestHelpers.DoubleEchoAgent("agent1"), new OrchestrationTestHelpers.DoubleEchoAgent("agent2")) + .AddParticipants(new OrchestrationTestHelpers.DoubleEchoAgent("agent3")) .Build(); for (int iter = 0; iter < 3; iter++) { const string UserInput = "abc"; - (string updateText, List? result, _, _) = await AgentWorkflowBuilderTests.RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, UserInput)]); + (string updateText, List? result, _, _) = await OrchestrationTestHelpers.RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, UserInput)]); Assert.NotNull(result); Assert.Equal(maxIterations + 1, result.Count); @@ -166,9 +166,9 @@ public class GroupChatWorkflowBuilderTests [Fact] public void Test_GroupChatWorkflowBuilder_DefaultDesignationsMatchSpec() { - AgentWorkflowBuilderTests.DoubleEchoAgent a1 = new("agent1"); - AgentWorkflowBuilderTests.DoubleEchoAgent a2 = new("agent2"); - AgentWorkflowBuilderTests.DoubleEchoAgent a3 = new("agent3"); + OrchestrationTestHelpers.DoubleEchoAgent a1 = new("agent1"); + OrchestrationTestHelpers.DoubleEchoAgent a2 = new("agent2"); + OrchestrationTestHelpers.DoubleEchoAgent a3 = new("agent3"); Workflow workflow = AgentWorkflowBuilder .CreateGroupChatBuilderWith(agents => new RoundRobinGroupChatManager(agents) { MaximumIterationCount = 1 }) @@ -186,9 +186,9 @@ public class GroupChatWorkflowBuilderTests [Fact] public void Test_GroupChatWorkflowBuilder_ExplicitDesignationsReplaceDefaults() { - AgentWorkflowBuilderTests.DoubleEchoAgent a1 = new("agent1"); - AgentWorkflowBuilderTests.DoubleEchoAgent a2 = new("agent2"); - AgentWorkflowBuilderTests.DoubleEchoAgent a3 = new("agent3"); + OrchestrationTestHelpers.DoubleEchoAgent a1 = new("agent1"); + OrchestrationTestHelpers.DoubleEchoAgent a2 = new("agent2"); + OrchestrationTestHelpers.DoubleEchoAgent a3 = new("agent3"); Workflow workflow = AgentWorkflowBuilder .CreateGroupChatBuilderWith(agents => new RoundRobinGroupChatManager(agents) { MaximumIterationCount = 1 }) @@ -210,8 +210,8 @@ public class GroupChatWorkflowBuilderTests [Fact] public void Test_GroupChatWorkflowBuilder_DesignationForNonParticipantThrows() { - AgentWorkflowBuilderTests.DoubleEchoAgent participant = new("p1"); - AgentWorkflowBuilderTests.DoubleEchoAgent stranger = new("stranger"); + OrchestrationTestHelpers.DoubleEchoAgent participant = new("p1"); + OrchestrationTestHelpers.DoubleEchoAgent stranger = new("stranger"); GroupChatWorkflowBuilder builder = AgentWorkflowBuilder .CreateGroupChatBuilderWith(agents => new RoundRobinGroupChatManager(agents) { MaximumIterationCount = 1 }) diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/HandoffWorkflowBuilderTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/HandoffWorkflowBuilderTests.cs index 92f4aa633d..c3b9eb6a90 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/HandoffWorkflowBuilderTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/HandoffWorkflowBuilderTests.cs @@ -19,8 +19,8 @@ public class HandoffWorkflowBuilderTests [Fact] public void Test_HandoffWorkflowBuilder_DefaultDesignationsMatchSpec() { - AgentWorkflowBuilderTests.DoubleEchoAgent coordinator = new("coordinator"); - AgentWorkflowBuilderTests.DoubleEchoAgent specialist = new("specialist"); + OrchestrationTestHelpers.DoubleEchoAgent coordinator = new("coordinator"); + OrchestrationTestHelpers.DoubleEchoAgent specialist = new("specialist"); Workflow workflow = AgentWorkflowBuilder .CreateHandoffBuilderWith(coordinator) @@ -38,8 +38,8 @@ public class HandoffWorkflowBuilderTests [Fact] public void Test_HandoffWorkflowBuilder_ExplicitDesignationsReplaceDefaults() { - AgentWorkflowBuilderTests.DoubleEchoAgent coordinator = new("coordinator"); - AgentWorkflowBuilderTests.DoubleEchoAgent specialist = new("specialist"); + OrchestrationTestHelpers.DoubleEchoAgent coordinator = new("coordinator"); + OrchestrationTestHelpers.DoubleEchoAgent specialist = new("specialist"); Workflow workflow = AgentWorkflowBuilder .CreateHandoffBuilderWith(coordinator) @@ -61,9 +61,9 @@ public class HandoffWorkflowBuilderTests [Fact] public void Test_HandoffWorkflowBuilder_DesignationForNonParticipantThrows() { - AgentWorkflowBuilderTests.DoubleEchoAgent coordinator = new("coordinator"); - AgentWorkflowBuilderTests.DoubleEchoAgent specialist = new("specialist"); - AgentWorkflowBuilderTests.DoubleEchoAgent stranger = new("stranger"); + OrchestrationTestHelpers.DoubleEchoAgent coordinator = new("coordinator"); + OrchestrationTestHelpers.DoubleEchoAgent specialist = new("specialist"); + OrchestrationTestHelpers.DoubleEchoAgent stranger = new("stranger"); HandoffWorkflowBuilder builder = AgentWorkflowBuilder .CreateHandoffBuilderWith(coordinator) diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/OrchestrationTestHelpers.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/OrchestrationTestHelpers.cs new file mode 100644 index 0000000000..1f6b56655d --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/OrchestrationTestHelpers.cs @@ -0,0 +1,131 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Runtime.CompilerServices; +using System.Text; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows.InProc; +using Microsoft.Extensions.AI; + +namespace Microsoft.Agents.AI.Workflows.UnitTests; + +/// +/// Container for shared test helpers used by every orchestration-builder test class — +/// the DoubleEchoAgent family and the RunWorkflow* methods. The actual +/// test methods live in per-builder files (SequentialWorkflowBuilderTests, +/// ConcurrentWorkflowBuilderTests, GroupChatWorkflowBuilderTests, etc.). +/// +public static class OrchestrationTestHelpers +{ + internal class DoubleEchoAgent(string name) : AIAgent + { + public override string Name => name; + + protected override ValueTask CreateSessionCoreAsync(CancellationToken cancellationToken = default) + => new(new DoubleEchoAgentSession()); + + protected override ValueTask DeserializeSessionCoreAsync(JsonElement serializedState, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default) + => new(new DoubleEchoAgentSession()); + + protected override ValueTask SerializeSessionCoreAsync(AgentSession session, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default) + => default; + + protected override Task RunCoreAsync( + IEnumerable messages, AgentSession? session = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default) => + throw new NotImplementedException(); + + protected override async IAsyncEnumerable RunCoreStreamingAsync( + IEnumerable messages, AgentSession? session = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + await Task.Yield(); + + var contents = messages.SelectMany(m => m.Contents).ToList(); + string id = Guid.NewGuid().ToString("N"); + yield return new AgentResponseUpdate(ChatRole.Assistant, this.Name) { AuthorName = this.Name, MessageId = id }; + yield return new AgentResponseUpdate(ChatRole.Assistant, contents) { AuthorName = this.Name, MessageId = id }; + yield return new AgentResponseUpdate(ChatRole.Assistant, contents) { AuthorName = this.Name, MessageId = id }; + } + } + + internal sealed class DoubleEchoAgentSession() : AgentSession(); + + internal sealed class DoubleEchoAgentWithBarrier(string name, StrongBox> barrier, StrongBox remaining) : DoubleEchoAgent(name) + { + protected override async IAsyncEnumerable RunCoreStreamingAsync( + IEnumerable messages, AgentSession? session = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + if (Interlocked.Decrement(ref remaining.Value) == 0) + { + barrier.Value!.SetResult(true); + } + + await barrier.Value!.Task.ConfigureAwait(false); + + await foreach (var update in base.RunCoreStreamingAsync(messages, session, options, cancellationToken)) + { + await Task.Yield(); + yield return update; + } + } + } + + internal sealed record WorkflowRunResult(string UpdateText, List? Result, CheckpointInfo? LastCheckpoint, List PendingRequests); + + internal static async Task RunWorkflowCheckpointedAsync( + Workflow workflow, List input, InProcessExecutionEnvironment environment, CheckpointInfo? fromCheckpoint = null) + { + await using StreamingRun run = + fromCheckpoint != null ? await environment.ResumeStreamingAsync(workflow, fromCheckpoint) + : await environment.OpenStreamingAsync(workflow); + + await run.TrySendMessageAsync(input); + await run.TrySendMessageAsync(new TurnToken(emitEvents: true)); + + return await ProcessWorkflowRunAsync(run); + } + + internal static async Task ProcessWorkflowRunAsync(StreamingRun run) + { + StringBuilder sb = new(); + WorkflowOutputEvent? output = null; + CheckpointInfo? lastCheckpoint = null; + + List pendingRequests = []; + + await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false).ConfigureAwait(false)) + { + switch (evt) + { + case AgentResponseUpdateEvent responseUpdate: + sb.Append(responseUpdate.Data); + break; + + case RequestInfoEvent requestInfo: + pendingRequests.Add(requestInfo); + break; + + case WorkflowOutputEvent e: + output = e; + break; + + case WorkflowErrorEvent errorEvent: + Assert.Fail($"Workflow execution failed with error: {errorEvent.Exception}"); + break; + + case SuperStepCompletedEvent stepCompleted: + lastCheckpoint = stepCompleted.CompletionInfo?.Checkpoint; + break; + } + } + + return new(sb.ToString(), output?.As>(), lastCheckpoint, pendingRequests); + } + + internal static Task RunWorkflowAsync( + Workflow workflow, List input, ExecutionEnvironment executionEnvironment = ExecutionEnvironment.InProcess_Lockstep) + => RunWorkflowCheckpointedAsync(workflow, input, executionEnvironment.ToWorkflowExecutionEnvironment()); +} \ No newline at end of file diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/SequentialWorkflowBuilderTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/SequentialWorkflowBuilderTests.cs new file mode 100644 index 0000000000..dd3ce4dcfa --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/SequentialWorkflowBuilderTests.cs @@ -0,0 +1,184 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using FluentAssertions; +using Microsoft.Agents.AI.Workflows.UnitTests.Futures; +using Microsoft.Extensions.AI; +using Xunit; + +namespace Microsoft.Agents.AI.Workflows.UnitTests; + +public class SequentialWorkflowBuilderTests +{ + [Fact] + public void Test_SequentialWorkflowBuilder_InvalidArguments_Throws() + { + Assert.Throws("agents", () => new SequentialWorkflowBuilder(null!)); + Assert.Throws("agents", () => new SequentialWorkflowBuilder().Build()); + + Assert.Throws("agents", () => AgentWorkflowBuilder.BuildSequential(workflowName: null!, null!)); + Assert.Throws("agents", () => AgentWorkflowBuilder.BuildSequential()); + Assert.Throws("agents", () => AgentWorkflowBuilder.CreateSequentialBuilderWith(null!)); + } + + [Theory] + [InlineData(1)] + [InlineData(2)] + [InlineData(3)] + [InlineData(4)] + [InlineData(5)] + public async Task Test_SequentialWorkflowBuilder_AgentsRunInOrderAsync(int numAgents) + { + var workflow = new SequentialWorkflowBuilder( + from i in Enumerable.Range(1, numAgents) + select new OrchestrationTestHelpers.DoubleEchoAgent($"agent{i}")) + .Build(); + + for (int iter = 0; iter < 3; iter++) + { + const string UserInput = "abc"; + (string updateText, List? result, _, _) = + await OrchestrationTestHelpers.RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, UserInput)]); + + Assert.NotNull(result); + Assert.Equal(numAgents + 1, result.Count); + + Assert.Equal(ChatRole.User, result[0].Role); + Assert.Null(result[0].AuthorName); + Assert.Equal(UserInput, result[0].Text); + + string[] texts = new string[numAgents + 1]; + texts[0] = UserInput; + string expectedTotal = string.Empty; + for (int i = 1; i < numAgents + 1; i++) + { + string id = $"agent{((i - 1) % numAgents) + 1}"; + texts[i] = $"{id}{Double(string.Concat(texts.Take(i)))}"; + Assert.Equal(ChatRole.Assistant, result[i].Role); + Assert.Equal(id, result[i].AuthorName); + Assert.Equal(texts[i], result[i].Text); + expectedTotal += texts[i]; + } + + Assert.Equal(expectedTotal, updateText); + Assert.Equal(UserInput + expectedTotal, string.Concat(result)); + + static string Double(string s) => s + s; + } + } + + [Fact] + public void Test_SequentialWorkflowBuilder_DefaultDesignationsMatchSpec() + { + Workflow workflow = new SequentialWorkflowBuilder( + new OrchestrationTestHelpers.DoubleEchoAgent("agent1"), + new OrchestrationTestHelpers.DoubleEchoAgent("agent2"), + new OrchestrationTestHelpers.DoubleEchoAgent("agent3")) + .Build(); + + Dictionary> designations = workflow.OutputExecutors; + designations.Where(kvp => kvp.Value.Count == 0) + .Should().ContainSingle("OutputMessagesExecutor is the sole terminal output by default"); + designations.Where(kvp => kvp.Value.Contains(OutputTag.Intermediate)) + .Should().HaveCount(3, "every pipeline agent is designated intermediate by default"); + } + + [Fact] + public void Test_SequentialWorkflowBuilder_ExplicitDesignationsReplaceDefaults() + { + OrchestrationTestHelpers.DoubleEchoAgent a1 = new("agent1"); + OrchestrationTestHelpers.DoubleEchoAgent a2 = new("agent2"); + OrchestrationTestHelpers.DoubleEchoAgent a3 = new("agent3"); + + Workflow workflow = new SequentialWorkflowBuilder(a1, a2, a3) + .WithOutputFrom(a1) + .WithIntermediateOutputFrom([a2]) + .Build(); + + Dictionary> designations = workflow.OutputExecutors; + + designations.Should().HaveCount(2, + "only the two explicitly-designated agents land on the inner builder; the end default is suppressed"); + designations.Values.Where(tags => tags.Count == 0) + .Should().ContainSingle("agent1 is the only terminal designation"); + designations.Values.Where(tags => tags.Contains(OutputTag.Intermediate)) + .Should().ContainSingle("agent2 is the only intermediate designation"); + } + + [Fact] + public void Test_SequentialWorkflowBuilder_DesignationForNonParticipantThrows() + { + OrchestrationTestHelpers.DoubleEchoAgent participant = new("p1"); + OrchestrationTestHelpers.DoubleEchoAgent stranger = new("stranger"); + + SequentialWorkflowBuilder builder = new SequentialWorkflowBuilder(participant) + .WithIntermediateOutputFrom([stranger]); + + Action build = () => builder.Build(); + build.Should().Throw().WithMessage("*stranger*"); + } + + [Fact] + public void Test_SequentialWorkflowBuilder_WithNamePropagatesToWorkflow() + { + Workflow workflow = new SequentialWorkflowBuilder(new OrchestrationTestHelpers.DoubleEchoAgent("agent1")) + .WithName("named-sequential") + .Build(); + + workflow.Name.Should().Be("named-sequential"); + } + + [Fact] + public void Test_SequentialWorkflowBuilder_WithDescriptionPropagatesToWorkflow() + { + Workflow workflow = new SequentialWorkflowBuilder(new OrchestrationTestHelpers.DoubleEchoAgent("agent1")) + .WithDescription("describes the sequential pipeline") + .Build(); + + workflow.Description.Should().Be("describes the sequential pipeline"); + } + + [Collection(FuturesSerialCollection.Name)] + public class AsAgentForwarding + { + [Fact] + public async Task Test_SequentialWorkflowBuilder_AsAgent_OnlyTerminalDesignationSurfacesAsync() + { + using FuturesScope _ = new(enabled: true); + + OrchestrationTestHelpers.DoubleEchoAgent agent1 = new("agent1"); + OrchestrationTestHelpers.DoubleEchoAgent agent2 = new("agent2"); + OrchestrationTestHelpers.DoubleEchoAgent agent3 = new("agent3"); + + // Explicitly designate ONLY the last agent — defaults (which would tag every agent + // intermediate) are suppressed, so under Futures-on, agent1/agent2 produce no + // AgentResponse(Update)Events and nothing of theirs reaches the AsAgent stream. + Workflow workflow = new SequentialWorkflowBuilder(agent1, agent2, agent3) + .WithOutputFrom(agent3) + .Build(); + + List updates = await workflow + .AsAIAgent("WorkflowAgent") + .RunStreamingAsync(new ChatMessage(ChatRole.User, "abc")) + .ToListAsync(); + + // Filter by AuthorName — distinguishes which agent originated each update + // (text-content checks are unreliable because agent3 echoes earlier agents' markers + // as part of the cumulative pipeline payload). + HashSet authoredBy = updates + .Select(u => u.AuthorName) + .Where(n => !string.IsNullOrEmpty(n)) + .Select(n => n!) + .ToHashSet(); + + authoredBy.Should().Contain("agent3", "the terminal agent must surface"); + authoredBy.Should().NotContain("agent1", + "the intermediate agent must not surface when only the terminal is designated"); + authoredBy.Should().NotContain("agent2", + "the intermediate agent must not surface when only the terminal is designated"); + } + } +}