diff --git a/dotnet/samples/GettingStarted/Workflows/Agents/WorkflowAsAnAgent/WorkflowFactory.cs b/dotnet/samples/GettingStarted/Workflows/Agents/WorkflowAsAnAgent/WorkflowFactory.cs index 736c51d555..653ebdf4c2 100644 --- a/dotnet/samples/GettingStarted/Workflows/Agents/WorkflowAsAnAgent/WorkflowFactory.cs +++ b/dotnet/samples/GettingStarted/Workflows/Agents/WorkflowAsAnAgent/WorkflowFactory.cs @@ -23,8 +23,8 @@ internal static class WorkflowFactory // Build the workflow by adding executors and connecting them return new WorkflowBuilder(startExecutor) - .AddFanOutEdge(startExecutor, targets: [frenchAgent, englishAgent]) - .AddFanInEdge(aggregationExecutor, sources: [frenchAgent, englishAgent]) + .AddFanOutEdge(startExecutor, [frenchAgent, englishAgent]) + .AddFanInEdge([frenchAgent, englishAgent], aggregationExecutor) .WithOutputFrom(aggregationExecutor) .Build(); } diff --git a/dotnet/samples/GettingStarted/Workflows/Concurrent/Concurrent/Program.cs b/dotnet/samples/GettingStarted/Workflows/Concurrent/Concurrent/Program.cs index e1f71e2311..c839149d6c 100644 --- a/dotnet/samples/GettingStarted/Workflows/Concurrent/Concurrent/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Concurrent/Concurrent/Program.cs @@ -52,8 +52,8 @@ public static class Program // Build the workflow by adding executors and connecting them var workflow = new WorkflowBuilder(startExecutor) - .AddFanOutEdge(startExecutor, targets: [physicist, chemist]) - .AddFanInEdge(aggregationExecutor, sources: [physicist, chemist]) + .AddFanOutEdge(startExecutor, [physicist, chemist]) + .AddFanInEdge([physicist, chemist], aggregationExecutor) .WithOutputFrom(aggregationExecutor) .Build(); diff --git a/dotnet/samples/GettingStarted/Workflows/Concurrent/MapReduce/Program.cs b/dotnet/samples/GettingStarted/Workflows/Concurrent/MapReduce/Program.cs index 9fd4b66e70..1b36b3eeb0 100644 --- a/dotnet/samples/GettingStarted/Workflows/Concurrent/MapReduce/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Concurrent/MapReduce/Program.cs @@ -62,10 +62,10 @@ public static class Program // Step 4: Build the concurrent workflow with fan-out/fan-in pattern return new WorkflowBuilder(splitter) - .AddFanOutEdge(splitter, targets: [.. mappers]) // Split -> many mappers - .AddFanInEdge(shuffler, sources: [.. mappers]) // All mappers -> shuffle - .AddFanOutEdge(shuffler, targets: [.. reducers]) // Shuffle -> many reducers - .AddFanInEdge(completion, sources: [.. reducers]) // All reducers -> completion + .AddFanOutEdge(splitter, [.. mappers]) // Split -> many mappers + .AddFanInEdge([.. mappers], shuffler) // All mappers -> shuffle + .AddFanOutEdge(shuffler, [.. reducers]) // Shuffle -> many reducers + .AddFanInEdge([.. reducers], completion) // All reducers -> completion .WithOutputFrom(completion) .Build(); } diff --git a/dotnet/samples/GettingStarted/Workflows/ConditionalEdges/03_MultiSelection/Program.cs b/dotnet/samples/GettingStarted/Workflows/ConditionalEdges/03_MultiSelection/Program.cs index 15746f727e..9d340cbae3 100644 --- a/dotnet/samples/GettingStarted/Workflows/ConditionalEdges/03_MultiSelection/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/ConditionalEdges/03_MultiSelection/Program.cs @@ -60,13 +60,13 @@ public static class Program WorkflowBuilder builder = new(emailAnalysisExecutor); builder.AddFanOutEdge( emailAnalysisExecutor, - targets: [ + [ handleSpamExecutor, emailAssistantExecutor, emailSummaryExecutor, handleUncertainExecutor, ], - partitioner: GetPartitioner() + GetTargetAssigner() ) // After the email assistant writes a response, it will be sent to the send email executor .AddEdge(emailAssistantExecutor, sendEmailExecutor) @@ -105,7 +105,7 @@ public static class Program /// Creates a partitioner for routing messages based on the analysis result. /// /// A function that takes an analysis result and returns the target partitions. - private static Func> GetPartitioner() + private static Func> GetTargetAssigner() { return (analysisResult, targetCount) => { diff --git a/dotnet/samples/GettingStarted/Workflows/Observability/WorkflowAsAnAgent/WorkflowHelper.cs b/dotnet/samples/GettingStarted/Workflows/Observability/WorkflowAsAnAgent/WorkflowHelper.cs index 816dce50d0..8069a3e88e 100644 --- a/dotnet/samples/GettingStarted/Workflows/Observability/WorkflowAsAnAgent/WorkflowHelper.cs +++ b/dotnet/samples/GettingStarted/Workflows/Observability/WorkflowAsAnAgent/WorkflowHelper.cs @@ -24,8 +24,8 @@ internal static class WorkflowHelper // Build the workflow by adding executors and connecting them return new WorkflowBuilder(startExecutor) - .AddFanOutEdge(startExecutor, targets: [frenchAgent, englishAgent]) - .AddFanInEdge(aggregationExecutor, sources: [frenchAgent, englishAgent]) + .AddFanOutEdge(startExecutor, [frenchAgent, englishAgent]) + .AddFanInEdge([frenchAgent, englishAgent], aggregationExecutor) .WithOutputFrom(aggregationExecutor) .Build(); } diff --git a/dotnet/samples/GettingStarted/Workflows/SharedStates/Program.cs b/dotnet/samples/GettingStarted/Workflows/SharedStates/Program.cs index 6f4cfdf38b..b7cbc25515 100644 --- a/dotnet/samples/GettingStarted/Workflows/SharedStates/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/SharedStates/Program.cs @@ -26,8 +26,8 @@ public static class Program // Build the workflow by connecting executors sequentially var workflow = new WorkflowBuilder(fileRead) - .AddFanOutEdge(fileRead, targets: [wordCount, paragraphCount]) - .AddFanInEdge(aggregate, sources: [wordCount, paragraphCount]) + .AddFanOutEdge(fileRead, [wordCount, paragraphCount]) + .AddFanInEdge([wordCount, paragraphCount], aggregate) .WithOutputFrom(aggregate) .Build(); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs index 81d254aa53..41f0d834f0 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs @@ -127,7 +127,7 @@ public static partial class AgentWorkflowBuilder // provenance tracking exposed in the workflow context passed to a handler. ExecutorBinding[] agentExecutors = (from agent in agents select (ExecutorBinding)new AgentRunStreamingExecutor(agent, includeInputInOutput: false)).ToArray(); ExecutorBinding[] accumulators = [.. from agent in agentExecutors select (ExecutorBinding)new CollectChatMessagesExecutor($"Batcher/{agent.Id}")]; - builder.AddFanOutEdge(start, targets: agentExecutors); + builder.AddFanOutEdge(start, agentExecutors); for (int i = 0; i < agentExecutors.Length; i++) { builder.AddEdge(agentExecutors[i], accumulators[i]); @@ -143,7 +143,7 @@ public static partial class AgentWorkflowBuilder ExecutorBinding end = endFactory.BindExecutor(ConcurrentEndExecutor.ExecutorId); - builder.AddFanInEdge(end, sources: accumulators); + builder.AddFanInEdge(accumulators, end); builder = builder.WithOutputFrom(end); if (workflowName is not null) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/SwitchBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/SwitchBuilder.cs index e180650935..b8cd6b6e78 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/SwitchBuilder.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/SwitchBuilder.cs @@ -84,7 +84,7 @@ public sealed class SwitchBuilder List<(Func Predicate, HashSet OutgoingIndicies)> caseMap = this._caseMap; HashSet defaultIndicies = this._defaultIndicies; - return builder.AddFanOutEdge(source, CasePartitioner, [.. this._executors]); + return builder.AddFanOutEdge(source, this._executors, CasePartitioner); IEnumerable CasePartitioner(object? input, int targetCount) { diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowBuilder.cs index c277a84b36..3e9ee0d143 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowBuilder.cs @@ -272,12 +272,12 @@ public class WorkflowBuilder /// The source executor from which the fan-out edge originates. Cannot be null. /// One or more target executors that will receive the fan-out edge. Cannot be null or empty. /// The current instance of . - public WorkflowBuilder AddFanOutEdge(ExecutorBinding source, params IEnumerable targets) - => this.AddFanOutEdge(source, null, targets); + public WorkflowBuilder AddFanOutEdge(ExecutorBinding source, IEnumerable targets) + => this.AddFanOutEdge(source, targets, null); - internal static Func>? CreateEdgeAssignerFunc(Func>? partitioner) + internal static Func>? CreateTargetAssignerFunc(Func>? targetAssigner) { - if (partitioner is null) + if (targetAssigner is null) { return null; } @@ -289,7 +289,7 @@ public class WorkflowBuilder maybeObj = portableValue.AsType(typeof(T)); } - return partitioner(maybeObj is T typed ? typed : default, count); + return targetAssigner(maybeObj is T typed ? typed : default, count); }; } @@ -300,11 +300,11 @@ public class WorkflowBuilder /// If a partitioner function is provided, it will be used to distribute input across the target /// executors. The order of targets determines their mapping in the partitioning process. /// The source executor from which the fan-out edge originates. Cannot be null. - /// An optional function that determines how input is partitioned among the target executors. - /// If null, messages will route to all targets. /// One or more target executors that will receive the fan-out edge. Cannot be null or empty. /// The current instance of . - public WorkflowBuilder AddFanOutEdge(ExecutorBinding source, Func>? partitioner = null, params IEnumerable targets) + /// An optional function that determines how input is assigned among the target executors. + /// If null, messages will route to all targets. + public WorkflowBuilder AddFanOutEdge(ExecutorBinding source, IEnumerable targets, Func>? targetSelector = null) { Throw.IfNull(source); Throw.IfNull(targets); @@ -321,7 +321,7 @@ public class WorkflowBuilder this.Track(source).Id, sinkIds, this.TakeEdgeId(), - CreateEdgeAssignerFunc(partitioner)); + CreateTargetAssignerFunc(targetSelector)); this.EnsureEdgesFor(source.Id).Add(new(fanOutEdge)); @@ -335,10 +335,10 @@ public class WorkflowBuilder /// This method establishes a fan-in relationship, allowing the target executor to be activated /// based on the completion or state of multiple sources. The trigger parameter can be used to customize activation /// behavior. - /// The target executor that receives input from the specified source executors. Cannot be null. /// One or more source executors that provide input to the target. Cannot be null or empty. + /// The target executor that receives input from the specified source executors. Cannot be null. /// The current instance of . - public WorkflowBuilder AddFanInEdge(ExecutorBinding target, params IEnumerable sources) + public WorkflowBuilder AddFanInEdge(IEnumerable sources, ExecutorBinding target) { Throw.IfNull(target); Throw.IfNull(sources); @@ -364,8 +364,14 @@ public class WorkflowBuilder return this; } + /// + [Obsolete("Use AddFanInEdge(IEnumerable, ExecutorBinding) instead.")] + public WorkflowBuilder AddFanInEdge(ExecutorBinding target, params IEnumerable sources) + => this.AddFanInEdge(sources, target); + private void Validate() { + // Validate that there are no unbound executors if (this._unboundExecutors.Count > 0) { throw new InvalidOperationException( diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowBuilderExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowBuilderExtensions.cs index e338f73fc6..c702cf9ece 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowBuilderExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowBuilderExtensions.cs @@ -22,10 +22,10 @@ public static class WorkflowBuilderExtensions /// The type of message to forward. /// The to which the edges will be added. /// The source executor from which messages will be forwarded. - /// The target executors to which messages will be forwarded. + /// The target executor to which messages will be forwarded. /// The updated instance. - public static WorkflowBuilder ForwardMessage(this WorkflowBuilder builder, ExecutorBinding source, params IEnumerable executors) - => builder.ForwardMessage(source, condition: null, executors); + public static WorkflowBuilder ForwardMessage(this WorkflowBuilder builder, ExecutorBinding source, ExecutorBinding target) + => builder.ForwardMessage(source, [target], condition: null); /// /// Adds edges to the workflow that forward messages of the specified type from the source executor to @@ -34,26 +34,38 @@ public static class WorkflowBuilderExtensions /// The type of message to forward. /// The to which the edges will be added. /// The source executor from which messages will be forwarded. + /// The target executors to which messages will be forwarded. + /// The updated instance. + public static WorkflowBuilder ForwardMessage(this WorkflowBuilder builder, ExecutorBinding source, IEnumerable targets) + => builder.ForwardMessage(source, targets, condition: null); + + /// + /// Adds edges to the workflow that forward messages of the specified type from the source executor to + /// one or more target executors. + /// + /// The type of message to forward. + /// The to which the edges will be added. + /// The source executor from which messages will be forwarded. + /// The target executors to which messages will be forwarded. /// An optional condition that messages must satisfy to be forwarded. If , /// all messages of type will be forwarded. - /// The target executors to which messages will be forwarded. /// The updated instance. - public static WorkflowBuilder ForwardMessage(this WorkflowBuilder builder, ExecutorBinding source, Func? condition = null, params IEnumerable executors) + public static WorkflowBuilder ForwardMessage(this WorkflowBuilder builder, ExecutorBinding source, IEnumerable targets, Func? condition = null) { - Throw.IfNull(executors); + Throw.IfNull(targets); Func predicate = WorkflowBuilder.CreateConditionFunc(IsAllowedTypeAndMatchingCondition)!; #if NET - if (executors.TryGetNonEnumeratedCount(out int count) && count == 1) + if (targets.TryGetNonEnumeratedCount(out int count) && count == 1) #else - if (executors is ICollection { Count: 1 }) + if (targets is ICollection { Count: 1 }) #endif { - return builder.AddEdge(source, executors.First(), predicate); + return builder.AddEdge(source, targets.First(), predicate); } - return builder.AddSwitch(source, (switch_) => switch_.AddCase(predicate, executors)); + return builder.AddSwitch(source, (switch_) => switch_.AddCase(predicate, targets)); // The reason we can check for "not null" here is that CreateConditionFunc will do the correct unwrapping // logic for PortableValues. @@ -66,24 +78,35 @@ public static class WorkflowBuilderExtensions /// The type of messages to exclude from being forwarded to the executors. /// The instance to which the edges will be added. /// The source executor from which messages will be forwarded. - /// The target executors to which messages, except those of type , will be forwarded. + /// The target executor to which messages, except those of type , will be forwarded. /// The updated instance with the added edges. - public static WorkflowBuilder ForwardExcept(this WorkflowBuilder builder, ExecutorBinding source, params IEnumerable executors) + public static WorkflowBuilder ForwardExcept(this WorkflowBuilder builder, ExecutorBinding source, ExecutorBinding target) + => builder.ForwardExcept(source, [target]); + + /// + /// Adds edges from the specified source to the provided executors, excluding messages of a specified type. + /// + /// The type of messages to exclude from being forwarded to the executors. + /// The instance to which the edges will be added. + /// The source executor from which messages will be forwarded. + /// The target executors to which messages, except those of type , will be forwarded. + /// The updated instance with the added edges. + public static WorkflowBuilder ForwardExcept(this WorkflowBuilder builder, ExecutorBinding source, IEnumerable targets) { - Throw.IfNull(executors); + Throw.IfNull(targets); Func predicate = WorkflowBuilder.CreateConditionFunc((Func)IsAllowedType)!; #if NET - if (executors.TryGetNonEnumeratedCount(out int count) && count == 1) + if (targets.TryGetNonEnumeratedCount(out int count) && count == 1) #else - if (executors is ICollection { Count: 1 }) + if (targets is ICollection { Count: 1 }) #endif { - return builder.AddEdge(source, executors.First(), predicate); + return builder.AddEdge(source, targets.First(), predicate); } - return builder.AddSwitch(source, (switch_) => switch_.AddCase(predicate, executors)); + return builder.AddSwitch(source, (switch_) => switch_.AddCase(predicate, targets)); // The reason we can check for "null" here is that CreateConditionFunc will do the correct unwrapping // logic for PortableValues. @@ -98,11 +121,11 @@ public static class WorkflowBuilderExtensions /// executor in the order provided. /// The workflow builder to which the executor chain will be added. /// The initial executor in the chain. Cannot be null. - /// If set to , the same executor can be added to the chain multiple times. - /// An ordered array of executors to be added to the chain after the source. + /// An ordered sequence of executors to be added to the chain after the source. /// The original workflow builder instance with the specified executor chain added. + /// If set to , the same executor can be added to the chain multiple times. /// Thrown if there is a cycle in the chain. - public static WorkflowBuilder AddChain(this WorkflowBuilder builder, ExecutorBinding source, bool allowRepetition = false, params IEnumerable executors) + public static WorkflowBuilder AddChain(this WorkflowBuilder builder, ExecutorBinding source, IList executors, bool allowRepetition = false) { Throw.IfNull(builder); Throw.IfNull(source); diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/09_Subworkflow_ExternalRequest.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/09_Subworkflow_ExternalRequest.cs index aac3ed1d94..9173304a57 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/09_Subworkflow_ExternalRequest.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/09_Subworkflow_ExternalRequest.cs @@ -68,10 +68,10 @@ internal static class Step9EntryPoint var requestPort = RequestPort.Create(id); - return builder.ForwardMessage(source, executors: [filter], condition: message => message.DataIs()) - .ForwardMessage(filter, executors: [requestPort], condition: message => message.DataIs()) - .ForwardMessage(requestPort, executors: [filter], condition: message => message.DataIs()) - .ForwardMessage(filter, executors: [source], condition: message => message.DataIs()); + return builder.ForwardMessage(source, targets: [filter], condition: message => message.DataIs()) + .ForwardMessage(filter, targets: [requestPort], condition: message => message.DataIs()) + .ForwardMessage(requestPort, targets: [filter], condition: message => message.DataIs()) + .ForwardMessage(filter, targets: [source], condition: message => message.DataIs()); } public static WorkflowBuilder AddExternalRequest(this WorkflowBuilder builder, ExecutorBinding source, string? id = null) @@ -88,10 +88,10 @@ internal static class Step9EntryPoint public static WorkflowBuilder AddExternalRequest(this WorkflowBuilder builder, ExecutorBinding source, RequestPort inputPort) { - return builder.ForwardMessage(source, inputPort) - .ForwardMessage(source, inputPort) - .ForwardMessage(inputPort, source) - .ForwardMessage(inputPort, source); + return builder.ForwardMessage(source, [inputPort]) + .ForwardMessage(source, [inputPort]) + .ForwardMessage(inputPort, [source]) + .ForwardMessage(inputPort, [source]); } public static Workflow CreateSubWorkflow() @@ -113,7 +113,7 @@ internal static class Step9EntryPoint ExecutorBinding subworkflow = CreateSubWorkflow().BindAsExecutor("ResourceWorkflow"); return new WorkflowBuilder(coordinator) - .AddChain(coordinator, allowRepetition: true, subworkflow, coordinator) + .AddChain(coordinator, [subworkflow, coordinator], allowRepetition: true) .AddPassthroughRequestHandler(subworkflow, cache) .AddPassthroughRequestHandler(subworkflow, policyEngine) .WithOutputFrom(coordinator) diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowVisualizerTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowVisualizerTests.cs index 441dd28439..4e7aa51ea0 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowVisualizerTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowVisualizerTests.cs @@ -111,8 +111,8 @@ public class WorkflowVisualizerTests // Build a connected workflow: start fans out to s1 and s2, which then fan-in to t var workflow = new WorkflowBuilder("start") - .AddFanOutEdge(start, s1, s2) - .AddFanInEdge(t, s1, s2) // AddFanInEdge(target, sources) + .AddFanOutEdge(start, [s1, s2]) + .AddFanInEdge([s1, s2], t) // AddFanInEdge(target, sources) .Build(); var dotContent = workflow.ToDotString(); @@ -174,7 +174,7 @@ public class WorkflowVisualizerTests var target3 = new MockExecutor("target3"); var workflow = new WorkflowBuilder("start") - .AddFanOutEdge(start, target1, target2, target3) + .AddFanOutEdge(start, [target1, target2, target3]) .Build(); var dotContent = workflow.ToDotString(); @@ -199,8 +199,8 @@ public class WorkflowVisualizerTests var workflow = new WorkflowBuilder("start") .AddEdge(start, a, Condition) // Conditional edge - .AddFanOutEdge(a, b, c) // Fan-out - .AddFanInEdge(end, b, c) // Fan-in - AddFanInEdge(target, sources) + .AddFanOutEdge(a, [b, c]) // Fan-out + .AddFanInEdge([b, c], end) // Fan-in - AddFanInEdge(target, sources) .Build(); var dotContent = workflow.ToDotString(); @@ -307,8 +307,8 @@ public class WorkflowVisualizerTests var t = new ListStrTargetExecutor("t"); var workflow = new WorkflowBuilder("start") - .AddFanOutEdge(start, s1, s2) - .AddFanInEdge(t, s1, s2) + .AddFanOutEdge(start, [s1, s2]) + .AddFanInEdge([s1, s2], t) .Build(); var mermaidContent = workflow.ToMermaidString(); @@ -378,8 +378,8 @@ public class WorkflowVisualizerTests var workflow = new WorkflowBuilder("start") .AddEdge(start, a, Condition) // Conditional edge - .AddFanOutEdge(a, b, c) // Fan-out - .AddFanInEdge(end, b, c) // Fan-in + .AddFanOutEdge(a, [b, c]) // Fan-out + .AddFanInEdge([b, c], end) // Fan-in .Build(); var mermaidContent = workflow.ToMermaidString();