diff --git a/dotnet/samples/AgentWebChat/AgentWebChat.AgentHost/Program.cs b/dotnet/samples/AgentWebChat/AgentWebChat.AgentHost/Program.cs index 58fe403c8d..4eeb381585 100644 --- a/dotnet/samples/AgentWebChat/AgentWebChat.AgentHost/Program.cs +++ b/dotnet/samples/AgentWebChat/AgentWebChat.AgentHost/Program.cs @@ -60,10 +60,7 @@ builder.AddAIAgent("knights-and-knaves", (sp, key) => If the user asks a general question about their surrounding, make something up which is consistent with the scenario. """, "Narrator"); - // TODO: How to avoid sync-over-async here? -#pragma warning disable VSTHRD002 // Avoid problematic synchronous waits - return AgentWorkflowBuilder.BuildConcurrent([knight, knave, narrator]).AsAgentAsync(name: key).AsTask().GetAwaiter().GetResult(); -#pragma warning restore VSTHRD002 + return AgentWorkflowBuilder.BuildConcurrent([knight, knave, narrator]).AsAgent(name: key); }); // Workflow consisting of multiple specialized agents diff --git a/dotnet/samples/GettingStarted/Workflows/Agents/CustomAgentExecutors/Program.cs b/dotnet/samples/GettingStarted/Workflows/Agents/CustomAgentExecutors/Program.cs index be345b4656..5d5369883c 100644 --- a/dotnet/samples/GettingStarted/Workflows/Agents/CustomAgentExecutors/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Agents/CustomAgentExecutors/Program.cs @@ -48,7 +48,7 @@ public static class Program .Build(); // Execute the workflow - await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, "Create a slogan for a new electric SUV that is affordable and fun to drive."); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, input: "Create a slogan for a new electric SUV that is affordable and fun to drive."); await foreach (WorkflowEvent evt in run.WatchStreamAsync()) { if (evt is SloganGeneratedEvent or FeedbackEvent) diff --git a/dotnet/samples/GettingStarted/Workflows/Agents/WorkflowAsAnAgent/Program.cs b/dotnet/samples/GettingStarted/Workflows/Agents/WorkflowAsAnAgent/Program.cs index 16d4e57e69..e6bbdf7456 100644 --- a/dotnet/samples/GettingStarted/Workflows/Agents/WorkflowAsAnAgent/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Agents/WorkflowAsAnAgent/Program.cs @@ -35,7 +35,7 @@ public static class Program var chatClient = new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential()).GetChatClient(deploymentName).AsIChatClient(); // Create the workflow and turn it into an agent - var workflow = await WorkflowHelper.GetWorkflowAsync(chatClient); + var workflow = WorkflowHelper.GetWorkflow(chatClient); var agent = workflow.AsAgent("workflow-agent", "Workflow Agent"); var thread = agent.GetNewThread(); diff --git a/dotnet/samples/GettingStarted/Workflows/Agents/WorkflowAsAnAgent/WorkflowHelper.cs b/dotnet/samples/GettingStarted/Workflows/Agents/WorkflowAsAnAgent/WorkflowHelper.cs index 82ebffa050..3689f20bb8 100644 --- a/dotnet/samples/GettingStarted/Workflows/Agents/WorkflowAsAnAgent/WorkflowHelper.cs +++ b/dotnet/samples/GettingStarted/Workflows/Agents/WorkflowAsAnAgent/WorkflowHelper.cs @@ -13,7 +13,7 @@ internal static class WorkflowHelper /// /// The chat client to use for the agents /// A workflow that processes input using two language agents - internal static ValueTask>> GetWorkflowAsync(IChatClient chatClient) + internal static Workflow GetWorkflow(IChatClient chatClient) { // Create executors var startExecutor = new ConcurrentStartExecutor(); @@ -26,7 +26,7 @@ internal static class WorkflowHelper .AddFanOutEdge(startExecutor, targets: [frenchAgent, englishAgent]) .AddFanInEdge(aggregationExecutor, sources: [frenchAgent, englishAgent]) .WithOutputFrom(aggregationExecutor) - .BuildAsync>(); + .Build(); } /// diff --git a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndRehydrate/Program.cs b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndRehydrate/Program.cs index aedf37700d..644869f1b3 100644 --- a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndRehydrate/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndRehydrate/Program.cs @@ -25,7 +25,7 @@ public static class Program private static async Task Main() { // Create the workflow - var workflow = await WorkflowHelper.GetWorkflowAsync(); + var workflow = WorkflowHelper.GetWorkflow(); // Create checkpoint manager var checkpointManager = CheckpointManager.Default; @@ -67,7 +67,7 @@ public static class Program Console.WriteLine($"Number of checkpoints created: {checkpoints.Count}"); // Rehydrate a new workflow instance from a saved checkpoint and continue execution - var newWorkflow = await WorkflowHelper.GetWorkflowAsync(); + var newWorkflow = WorkflowHelper.GetWorkflow(); const int CheckpointIndex = 5; Console.WriteLine($"\n\nHydrating a new workflow instance from the {CheckpointIndex + 1}th checkpoint."); CheckpointInfo savedCheckpoint = checkpoints[CheckpointIndex]; diff --git a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndRehydrate/WorkflowHelper.cs b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndRehydrate/WorkflowHelper.cs index bf38f74ec8..89b9b70a39 100644 --- a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndRehydrate/WorkflowHelper.cs +++ b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndRehydrate/WorkflowHelper.cs @@ -13,7 +13,7 @@ internal static class WorkflowHelper /// 2. JudgeExecutor: Evaluates the guess and provides feedback. /// The workflow continues until the correct number is guessed. /// - internal static ValueTask> GetWorkflowAsync() + internal static Workflow GetWorkflow() { // Create the executors GuessNumberExecutor guessNumberExecutor = new(1, 100); @@ -24,7 +24,7 @@ internal static class WorkflowHelper .AddEdge(guessNumberExecutor, judgeExecutor) .AddEdge(judgeExecutor, guessNumberExecutor) .WithOutputFrom(judgeExecutor) - .BuildAsync(); + .Build(); } } diff --git a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndResume/Program.cs b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndResume/Program.cs index b4191b397f..44cf3da2ee 100644 --- a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndResume/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndResume/Program.cs @@ -24,7 +24,7 @@ public static class Program private static async Task Main() { // Create the workflow - var workflow = await WorkflowHelper.GetWorkflowAsync(); + var workflow = WorkflowHelper.GetWorkflow(); // Create checkpoint manager var checkpointManager = CheckpointManager.Default; diff --git a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndResume/WorkflowHelper.cs b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndResume/WorkflowHelper.cs index 5f60b355d1..489a811293 100644 --- a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndResume/WorkflowHelper.cs +++ b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointAndResume/WorkflowHelper.cs @@ -13,7 +13,7 @@ internal static class WorkflowHelper /// 2. JudgeExecutor: Evaluates the guess and provides feedback. /// The workflow continues until the correct number is guessed. /// - internal static ValueTask> GetWorkflowAsync() + internal static Workflow GetWorkflow() { // Create the executors GuessNumberExecutor guessNumberExecutor = new(1, 100); @@ -24,7 +24,7 @@ internal static class WorkflowHelper .AddEdge(guessNumberExecutor, judgeExecutor) .AddEdge(judgeExecutor, guessNumberExecutor) .WithOutputFrom(judgeExecutor) - .BuildAsync(); + .Build(); } } diff --git a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointWithHumanInTheLoop/Program.cs b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointWithHumanInTheLoop/Program.cs index 0a968ed8b3..6fdf383001 100644 --- a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointWithHumanInTheLoop/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointWithHumanInTheLoop/Program.cs @@ -27,7 +27,7 @@ public static class Program private static async Task Main() { // Create the workflow - var workflow = await WorkflowHelper.GetWorkflowAsync(); + var workflow = WorkflowHelper.GetWorkflow(); // Create checkpoint manager var checkpointManager = CheckpointManager.Default; diff --git a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointWithHumanInTheLoop/WorkflowHelper.cs b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointWithHumanInTheLoop/WorkflowHelper.cs index a4dcfcf376..69bf31d1ce 100644 --- a/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointWithHumanInTheLoop/WorkflowHelper.cs +++ b/dotnet/samples/GettingStarted/Workflows/Checkpoint/CheckpointWithHumanInTheLoop/WorkflowHelper.cs @@ -10,7 +10,7 @@ internal static class WorkflowHelper /// Get a workflow that plays a number guessing game with human-in-the-loop interaction. /// An input port allows the external world to provide inputs to the workflow upon requests. /// - internal static ValueTask> GetWorkflowAsync() + internal static Workflow GetWorkflow() { // Create the executors RequestPort numberRequest = RequestPort.Create("GuessNumber"); @@ -21,7 +21,7 @@ internal static class WorkflowHelper .AddEdge(numberRequest, judgeExecutor) .AddEdge(judgeExecutor, numberRequest) .WithOutputFrom(judgeExecutor) - .BuildAsync(); + .Build(); } } diff --git a/dotnet/samples/GettingStarted/Workflows/Concurrent/Concurrent/Program.cs b/dotnet/samples/GettingStarted/Workflows/Concurrent/Concurrent/Program.cs index 30b2372006..dc91338987 100644 --- a/dotnet/samples/GettingStarted/Workflows/Concurrent/Concurrent/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Concurrent/Concurrent/Program.cs @@ -58,7 +58,7 @@ public static class Program .Build(); // Execute the workflow in streaming mode - await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, "What is temperature?"); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, input: "What is temperature?"); await foreach (WorkflowEvent evt in run.WatchStreamAsync()) { if (evt is WorkflowOutputEvent output) diff --git a/dotnet/samples/GettingStarted/Workflows/Concurrent/MapReduce/Program.cs b/dotnet/samples/GettingStarted/Workflows/Concurrent/MapReduce/Program.cs index db10f0e8af..9fd4b66e70 100644 --- a/dotnet/samples/GettingStarted/Workflows/Concurrent/MapReduce/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Concurrent/MapReduce/Program.cs @@ -99,7 +99,7 @@ public static class Program // Step 2: Run the workflow Console.WriteLine("\n=== RUNNING WORKFLOW ===\n"); - await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, rawText); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, input: rawText); await foreach (WorkflowEvent evt in run.WatchStreamAsync()) { Console.WriteLine($"Event: {evt}"); diff --git a/dotnet/samples/GettingStarted/Workflows/Declarative/ExecuteCode/Program.cs b/dotnet/samples/GettingStarted/Workflows/Declarative/ExecuteCode/Program.cs index 7837194d32..c1846dac5e 100644 --- a/dotnet/samples/GettingStarted/Workflows/Declarative/ExecuteCode/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Declarative/ExecuteCode/Program.cs @@ -44,7 +44,7 @@ internal sealed class Program // Run the workflow, just like any other workflow string input = this.GetWorkflowInput(); - StreamingRun run = await InProcessExecution.StreamAsync(workflow, input); + StreamingRun run = await InProcessExecution.StreamAsync(workflow, input: input); await this.MonitorAndDisposeWorkflowRunAsync(run); Notify("\nWORKFLOW: Done!"); diff --git a/dotnet/samples/GettingStarted/Workflows/HumanInTheLoop/HumanInTheLoopBasic/Program.cs b/dotnet/samples/GettingStarted/Workflows/HumanInTheLoop/HumanInTheLoopBasic/Program.cs index 62f1925fb3..db18062c35 100644 --- a/dotnet/samples/GettingStarted/Workflows/HumanInTheLoop/HumanInTheLoopBasic/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/HumanInTheLoop/HumanInTheLoopBasic/Program.cs @@ -24,7 +24,7 @@ public static class Program private static async Task Main() { // Create the workflow - var workflow = await WorkflowHelper.GetWorkflowAsync(); + var workflow = WorkflowHelper.GetWorkflow(); // Execute the workflow await using StreamingRun handle = await InProcessExecution.StreamAsync(workflow, NumberSignal.Init); diff --git a/dotnet/samples/GettingStarted/Workflows/HumanInTheLoop/HumanInTheLoopBasic/WorkflowHelper.cs b/dotnet/samples/GettingStarted/Workflows/HumanInTheLoop/HumanInTheLoopBasic/WorkflowHelper.cs index c87ddc00cb..057b42f52c 100644 --- a/dotnet/samples/GettingStarted/Workflows/HumanInTheLoop/HumanInTheLoopBasic/WorkflowHelper.cs +++ b/dotnet/samples/GettingStarted/Workflows/HumanInTheLoop/HumanInTheLoopBasic/WorkflowHelper.cs @@ -10,7 +10,7 @@ internal static class WorkflowHelper /// Get a workflow that plays a number guessing game with human-in-the-loop interaction. /// An input port allows the external world to provide inputs to the workflow upon requests. /// - internal static ValueTask> GetWorkflowAsync() + internal static Workflow GetWorkflow() { // Create the executors RequestPort numberRequestPort = RequestPort.Create("GuessNumber"); @@ -21,7 +21,7 @@ internal static class WorkflowHelper .AddEdge(numberRequestPort, judgeExecutor) .AddEdge(judgeExecutor, numberRequestPort) .WithOutputFrom(judgeExecutor) - .BuildAsync(); + .Build(); } } diff --git a/dotnet/samples/GettingStarted/Workflows/Loop/Program.cs b/dotnet/samples/GettingStarted/Workflows/Loop/Program.cs index 15d930216f..a4004f333e 100644 --- a/dotnet/samples/GettingStarted/Workflows/Loop/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Loop/Program.cs @@ -25,11 +25,11 @@ public static class Program JudgeExecutor judgeExecutor = new("Judge", 42); // Build the workflow by connecting executors in a loop - var workflow = await new WorkflowBuilder(guessNumberExecutor) + var workflow = new WorkflowBuilder(guessNumberExecutor) .AddEdge(guessNumberExecutor, judgeExecutor) .AddEdge(judgeExecutor, guessNumberExecutor) .WithOutputFrom(judgeExecutor) - .BuildAsync(); + .Build(); // Execute the workflow await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, NumberSignal.Init); diff --git a/dotnet/samples/GettingStarted/Workflows/_Foundational/02_Streaming/Program.cs b/dotnet/samples/GettingStarted/Workflows/_Foundational/02_Streaming/Program.cs index 021f191a5e..3406e361ff 100644 --- a/dotnet/samples/GettingStarted/Workflows/_Foundational/02_Streaming/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/_Foundational/02_Streaming/Program.cs @@ -28,7 +28,7 @@ public static class Program var workflow = builder.Build(); // Execute the workflow in streaming mode - await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, "Hello, World!"); + await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, input: "Hello, World!"); await foreach (WorkflowEvent evt in run.WatchStreamAsync()) { if (evt is ExecutorCompletedEvent executorCompleted) diff --git a/dotnet/samples/GettingStarted/Workflows/_Foundational/05_MultiModelService/Program.cs b/dotnet/samples/GettingStarted/Workflows/_Foundational/05_MultiModelService/Program.cs index 4ca7289439..c90131a27c 100644 --- a/dotnet/samples/GettingStarted/Workflows/_Foundational/05_MultiModelService/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/_Foundational/05_MultiModelService/Program.cs @@ -57,8 +57,7 @@ AIAgent reporter = new ChatClientAgent(anthropic, description: "Summarize the researcher's essay into a single paragraph, focusing only on the fact checker's confirmed facts."); // Build a sequential workflow: Researcher -> Fact-Checker -> Reporter -AIAgent workflowAgent = await AgentWorkflowBuilder.BuildSequential(researcher, factChecker, reporter) - .AsAgentAsync(); +AIAgent workflowAgent = AgentWorkflowBuilder.BuildSequential(researcher, factChecker, reporter).AsAgent(); // Run the workflow, streaming the output as it arrives. string? lastAuthor = null; diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting/HostedWorkflowBuilderExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Hosting/HostedWorkflowBuilderExtensions.cs index 26104c9a57..f303415255 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting/HostedWorkflowBuilderExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting/HostedWorkflowBuilderExtensions.cs @@ -27,12 +27,7 @@ public static class HostedWorkflowBuilderExtensions public static IHostedAgentBuilder AddAsAIAgent(this IHostedWorkflowBuilder builder, string? name) { var agentName = name ?? builder.Name; - return builder.HostApplicationBuilder.AddAIAgent(agentName, (sp, key) => - { - var workflow = sp.GetRequiredKeyedService(key); -#pragma warning disable VSTHRD002 // Avoid problematic synchronous waits - return workflow.AsAgentAsync(name: key).AsTask().GetAwaiter().GetResult(); -#pragma warning restore VSTHRD002 // Avoid problematic synchronous waits - }); + return builder.HostApplicationBuilder.AddAIAgent(agentName, (sp, key) => sp.GetRequiredKeyedService(key) + .AsAgent(name: key)); } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs index 907d10fe60..d4cdfcc452 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/AgentWorkflowBuilder.cs @@ -16,7 +16,7 @@ namespace Microsoft.Agents.AI.Workflows; public static partial class AgentWorkflowBuilder { /// - /// Builds a composed of a pipeline of agents where the output of one agent is the input to the next. + /// Builds a composed of a pipeline of agents where the output of one agent is the input to the next. /// /// The sequence of agents to compose into a sequential workflow. /// The built workflow composed of the supplied , in the order in which they were yielded from the source. @@ -24,7 +24,7 @@ public static partial class AgentWorkflowBuilder => BuildSequentialCore(workflowName: null, agents); /// - /// Builds a composed of a pipeline of agents where the output of one agent is the input to the next. + /// Builds a composed of a pipeline of agents where the output of one agent is the input to the next. /// /// The name of workflow. /// The sequence of agents to compose into a sequential workflow. @@ -76,7 +76,7 @@ public static partial class AgentWorkflowBuilder } /// - /// Builds a composed of agents that operate concurrently on the same input, + /// Builds a composed of agents that operate concurrently on the same input, /// aggregating their outputs into a single collection. /// /// The set of agents to compose into a concurrent workflow. @@ -92,7 +92,7 @@ public static partial class AgentWorkflowBuilder => BuildConcurrentCore(workflowName: null, agents, aggregator); /// - /// Builds a composed of agents that operate concurrently on the same input, + /// Builds a composed of agents that operate concurrently on the same input, /// aggregating their outputs into a single collection. /// /// The name of the workflow. diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/ChatProtocol.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/ChatProtocol.cs new file mode 100644 index 0000000000..ff8140ee3a --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/ChatProtocol.cs @@ -0,0 +1,61 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using Microsoft.Extensions.AI; + +namespace Microsoft.Agents.AI.Workflows; + +/// +/// Provides extension methods for determining and enforcing whether a protocol descriptor represents the Agent Workflow +/// Chat Protocol. +/// +/// This is defined as supporting a and as input. Optional support +/// for additional payloads (e.g. string, when a default role is defined), or other collections of +/// messages are optional to support. +/// +public static class ChatProtocolExtensions +{ + /// + /// Determines whether the specified protocol descriptor represents the Agent Workflow Chat Protocol. + /// + /// The protocol descriptor to evaluate. + /// if the protocol descriptor represents a supported chat protocol; otherwise, . + public static bool IsChatProtocol(this ProtocolDescriptor descriptor) + { + bool foundListChatMessageInput = false; + bool foundTurnTokenInput = false; + + // We require that the workflow be a ChatProtocol; right now that is defined as accepting at + // least List as input (pending polymorphism/interface-input support), as well as + // TurnToken. Since output is mediated by events, which we forward, we don't need to validate + // output type. + foreach (Type inputType in descriptor.Accepts) + { + if (inputType == typeof(List)) + { + foundListChatMessageInput = true; + } + else if (inputType == typeof(TurnToken)) + { + foundTurnTokenInput = true; + } + } + + return foundListChatMessageInput && foundTurnTokenInput; + } + + /// + /// Throws an exception if the specified protocol descriptor does not represent a valid chat protocol. + /// + /// The protocol descriptor to validate as a chat protocol. Cannot be null. + public static void ThrowIfNotChatProtocol(this ProtocolDescriptor descriptor) + { + if (!descriptor.IsChatProtocol()) + { + throw new InvalidOperationException("Workflow does not support ChatProtocol: At least List" + + " and TurnToken must be supported as input."); + } + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/RepresentationExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/RepresentationExtensions.cs index d22be4974d..1878cbdd21 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/RepresentationExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/RepresentationExtensions.cs @@ -33,7 +33,7 @@ internal static class RepresentationExtensions return new(new TypeId(port.Request), new TypeId(port.Response), port.Id); } - private static WorkflowInfo ToWorkflowInfo(this Workflow workflow, TypeId? inputType, TypeId? outputType, string? outputExecutorId) + public static WorkflowInfo ToWorkflowInfo(this Workflow workflow) { Throw.IfNull(workflow); @@ -48,12 +48,6 @@ internal static class RepresentationExtensions HashSet inputPorts = new(workflow.Ports.Values.Select(ToPortInfo)); - return new WorkflowInfo(executors, edges, inputPorts, inputType, workflow.StartExecutorId, workflow.OutputExecutors); + return new WorkflowInfo(executors, edges, inputPorts, workflow.StartExecutorId, workflow.OutputExecutors); } - - public static WorkflowInfo ToWorkflowInfo(this Workflow workflow) - => workflow.ToWorkflowInfo(inputType: null, outputType: null, outputExecutorId: null); - - public static WorkflowInfo ToWorkflowInfo(this Workflow workflow) - => workflow.ToWorkflowInfo(inputType: new(workflow.InputType), outputType: null, outputExecutorId: null); } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/WorkflowInfo.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/WorkflowInfo.cs index a63a316b23..59a41db405 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/WorkflowInfo.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Checkpointing/WorkflowInfo.cs @@ -14,7 +14,6 @@ internal sealed class WorkflowInfo Dictionary executors, Dictionary> edges, HashSet requestPorts, - TypeId? inputType, string startExecutorId, HashSet? outputExecutorIds) { @@ -22,7 +21,6 @@ internal sealed class WorkflowInfo this.Edges = Throw.IfNull(edges); this.RequestPorts = Throw.IfNull(requestPorts); - this.InputType = inputType; this.StartExecutorId = Throw.IfNullOrEmpty(startExecutorId); this.OutputExecutorIds = outputExecutorIds ?? []; } @@ -91,7 +89,4 @@ internal sealed class WorkflowInfo return true; } - - public bool IsMatch(Workflow workflow) => - this.IsMatch(workflow as Workflow) && this.InputType?.IsMatch() == true; } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Executor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Executor.cs index 97c7932cf0..4c2821476a 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Executor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Executor.cs @@ -200,6 +200,18 @@ public abstract class Executor : IIdentified /// public ISet OutputTypes { get; } = new HashSet([typeof(object)]); + /// + /// Describes the protocol for communication with this . + /// + /// + public ProtocolDescriptor DescribeProtocol() + { + // TODO: Once burden of annotating yield/output messages becomes easier for the non-Auto case, + // we should (1) start checking for validity on output/send side, and (2) add the Yield/Send + // types to the ProtocolDescriptor. + return new(this.InputTypes); + } + /// /// Checks if the executor can handle a specific message type. /// diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/ExecutorIsh.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/ExecutorIsh.cs index 2151dadce4..63807c5576 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/ExecutorIsh.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/ExecutorIsh.cs @@ -21,9 +21,8 @@ public static class ExecutorIshConfigurationExtensions /// Note that Executor Ids must be unique within a workflow. /// /// Although this will generally result in a delay-instantiated once messages are available - /// for it, if this is used as a start node of a typed via , - /// it will be instantiated as part of the workflow's construction, to validate that its input type matches the - /// demanded TInput. + /// for it, it will be instantiated if a for the is requested, + /// and it is the starting executor. /// /// The type of the resulting executor /// The factory method. @@ -38,9 +37,8 @@ public static class ExecutorIshConfigurationExtensions /// /// /// Although this will generally result in a delay-instantiated once messages are available - /// for it, if this is used as a start node of a typed via , - /// it will be instantiated as part of the workflow's construction, to validate that its input type matches the - /// demanded TInput. + /// for it, it will be instantiated if a for the is requested, + /// and it is the starting executor. /// /// The type of the resulting executor /// The factory method. @@ -56,9 +54,8 @@ public static class ExecutorIshConfigurationExtensions /// /// /// Although this will generally result in a delay-instantiated once messages are available - /// for it, if this is used as a start node of a typed via , - /// it will be instantiated as part of the workflow's construction, to validate that its input type matches the - /// demanded TInput. + /// for it, it will be instantiated if a for the is requested, + /// and it is the starting executor. /// /// The type of the resulting executor /// The type of options object to be passed to the factory method. diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/HandoffsWorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/HandoffsWorkflowBuilder.cs index 4362f0834f..9e5b61ac42 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/HandoffsWorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/HandoffsWorkflowBuilder.cs @@ -139,7 +139,7 @@ public sealed class HandoffsWorkflowBuilder } /// - /// Builds a composed of agents that operate via handoffs, with the next + /// Builds a composed of agents that operate via handoffs, with the next /// agent to process messages selected by the current agent. /// /// The workflow built based on the handoffs in the builder. diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowExecutionEnvironment.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowExecutionEnvironment.cs index cab1196e84..d1906e0df2 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowExecutionEnvironment.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowExecutionEnvironment.cs @@ -11,6 +11,16 @@ namespace Microsoft.Agents.AI.Workflows; /// public interface IWorkflowExecutionEnvironment { + /// + /// Initiates a streaming run of the specified workflow without sending any initial input. + /// + /// The workflow to execute. Cannot be null. + /// An optional identifier for the run. If null, a new run identifier will be generated. + /// A cancellation token that can be used to cancel the streaming operation. + /// A ValueTask that represents the asynchronous operation. The result contains a StreamingRun object for accessing + /// the streamed workflow output. + ValueTask StreamAsync(Workflow workflow, string? runId = null, CancellationToken cancellationToken = default); + /// /// Initiates an asynchronous streaming execution using the specified input. /// @@ -21,25 +31,24 @@ public interface IWorkflowExecutionEnvironment /// The workflow to be executed. Must not be null. /// The input message to be processed as part of the streaming run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . + /// The to monitor for cancellation requests. The default is . /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. ValueTask StreamAsync(Workflow workflow, TInput input, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; /// - /// Initiates an asynchronous streaming execution using the specified input. + /// Initiates an asynchronous streaming execution without sending any initial input, with checkpointing. /// /// The returned provides methods to observe and control /// the ongoing streaming execution. The operation will continue until the streaming execution is finished or /// cancelled. - /// A type of input accepted by the workflow. Must be non-nullable. /// The workflow to be executed. Must not be null. - /// The input message to be processed as part of the streaming run. + /// The to use with this run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . + /// The to monitor for cancellation requests. The default is . /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. - ValueTask StreamAsync(Workflow workflow, TInput input, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; + ValueTask> StreamAsync(Workflow workflow, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default); /// /// Initiates an asynchronous streaming execution using the specified input, with checkpointing. @@ -52,27 +61,11 @@ public interface IWorkflowExecutionEnvironment /// The input message to be processed as part of the streaming run. /// The to use with this run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . + /// The to monitor for cancellation requests. The default is . /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. ValueTask> StreamAsync(Workflow workflow, TInput input, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; - /// - /// Initiates an asynchronous streaming execution using the specified input, with checkpointing. - /// - /// The returned provides methods to observe and control - /// the ongoing streaming execution. The operation will continue until the streaming execution is finished or - /// cancelled. - /// The type of input accepted by the workflow. Must be non-nullable. - /// The workflow to be executed. Must not be null. - /// The input message to be processed as part of the streaming run. - /// The to use with this run. - /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . - /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. - ValueTask> StreamAsync(Workflow workflow, TInput input, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; - /// /// Resumes an asynchronous streaming execution for the specified input from a checkpoint. /// @@ -82,24 +75,10 @@ public interface IWorkflowExecutionEnvironment /// The corresponding to the checkpoint from which to resume. /// The to use with this run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . + /// The to monitor for cancellation requests. The default is . /// A that provides access to the results of the streaming run. ValueTask> ResumeStreamAsync(Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default); - /// - /// Resumes an asynchronous streaming execution for the specified input from a checkpoint. - /// - /// If the operation is cancelled via the token, the streaming execution will - /// be terminated. - /// The type of input accepted by the workflow. Must be non-nullable. - /// The workflow to be executed. Must not be null. - /// The corresponding to the checkpoint from which to resume. - /// The to use with this run. - /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . - /// A that provides access to the results of the streaming run. - ValueTask> ResumeStreamAsync(Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; - /// /// Initiates a non-streaming execution of the workflow with the specified input. /// @@ -109,25 +88,11 @@ public interface IWorkflowExecutionEnvironment /// The workflow to be executed. Must not be null. /// The input message to be processed as part of the run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . + /// The to monitor for cancellation requests. The default is . /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. ValueTask RunAsync(Workflow workflow, TInput input, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; - /// - /// Initiates a non-streaming execution of the workflow with the specified input. - /// - /// The workflow will run until its first halt, and the returned will capture - /// all outgoing events. Use the Run instance to resume execution with responses to outgoing events. - /// The type of input accepted by the workflow. Must be non-nullable. - /// The workflow to be executed. Must not be null. - /// The input message to be processed as part of the run. - /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . - /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. - ValueTask RunAsync(Workflow workflow, TInput input, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; - /// /// Initiates a non-streaming execution of the workflow with the specified input, with checkpointing. /// @@ -138,26 +103,11 @@ public interface IWorkflowExecutionEnvironment /// The input message to be processed as part of the run. /// The to use with this run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . + /// The to monitor for cancellation requests. The default is . /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. ValueTask> RunAsync(Workflow workflow, TInput input, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; - /// - /// Initiates a non-streaming execution of the workflow with the specified input, with checkpointing. - /// - /// The workflow will run until its first halt, and the returned will capture - /// all outgoing events. Use the Run instance to resume execution with responses to outgoing events. - /// The type of input accepted by the workflow. Must be non-nullable. - /// The workflow to be executed. Must not be null. - /// The input message to be processed as part of the run. - /// The to use with this run. - /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . - /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. - ValueTask> RunAsync(Workflow workflow, TInput input, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; - /// /// Resumes a non-streaming execution of the workflow from a checkpoint. /// @@ -167,22 +117,8 @@ public interface IWorkflowExecutionEnvironment /// The corresponding to the checkpoint from which to resume. /// The to use with this run. /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . + /// The to monitor for cancellation requests. The default is . /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. ValueTask> ResumeAsync(Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default); - - /// - /// Resumes a non-streaming execution of the workflow from a checkpoint. - /// - /// The workflow will run until its first halt, and the returned will capture - /// all outgoing events. Use the Run instance to resume execution with responses to outgoing events. - /// The workflow to be executed. Must not be null. - /// The corresponding to the checkpoint from which to resume. - /// The to use with this run. - /// An optional unique identifier for the run. If not provided, a new identifier will be generated. - /// The to monitor for cancellationToken requests. The default is . - /// A that represents the asynchronous operation. The result contains a for managing and interacting with the streaming run. - ValueTask> ResumeAsync(Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull; } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessExecutionEnvironment.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessExecutionEnvironment.cs index 219b4642cd..1bbed760db 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessExecutionEnvironment.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessExecutionEnvironment.cs @@ -36,6 +36,18 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen return runner.ResumeStreamAsync(this.ExecutionMode, fromCheckpoint, cancellationToken); } + /// + public async ValueTask StreamAsync( + Workflow workflow, + string? runId = null, + CancellationToken cancellationToken = default) + { + AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager: null, runId: runId, [], cancellationToken) + .ConfigureAwait(false); + + return new(runHandle); + } + /// public async ValueTask StreamAsync( Workflow workflow, @@ -50,16 +62,17 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen } /// - public async ValueTask StreamAsync( - Workflow workflow, - TInput input, + public async ValueTask> StreamAsync( + Workflow workflow, + CheckpointManager checkpointManager, string? runId = null, - CancellationToken cancellationToken = default) where TInput : notnull + CancellationToken cancellationToken = default) { - AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager: null, runId: runId, [typeof(TInput)], cancellationToken) + AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager, runId: runId, [], cancellationToken) .ConfigureAwait(false); - return await runHandle.EnqueueAndStreamAsync(input, cancellationToken).ConfigureAwait(false); + return await runHandle.WithCheckpointingAsync(() => new(new StreamingRun(runHandle))) + .ConfigureAwait(false); } /// @@ -77,21 +90,6 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen .ConfigureAwait(false); } - /// - public async ValueTask> StreamAsync( - Workflow workflow, - TInput input, - CheckpointManager checkpointManager, - string? runId = null, - CancellationToken cancellationToken = default) where TInput : notnull - { - AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager, runId: runId, [typeof(TInput)], cancellationToken) - .ConfigureAwait(false); - - return await runHandle.WithCheckpointingAsync(() => runHandle.EnqueueAndStreamAsync(input, cancellationToken)) - .ConfigureAwait(false); - } - /// public async ValueTask> ResumeStreamAsync( Workflow workflow, @@ -107,19 +105,24 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen .ConfigureAwait(false); } - /// - public async ValueTask> ResumeStreamAsync( - Workflow workflow, - CheckpointInfo fromCheckpoint, - CheckpointManager checkpointManager, + private async ValueTask BeginRunHandlingChatProtocolAsync(Workflow workflow, + TInput input, + CheckpointManager? checkpointManager, string? runId = null, - CancellationToken cancellationToken = default) where TInput : notnull + CancellationToken cancellationToken = default) { - AsyncRunHandle runHandle = await this.ResumeRunAsync(workflow, checkpointManager, runId: runId, fromCheckpoint, [typeof(TInput)], cancellationToken) + ProtocolDescriptor descriptor = await workflow.DescribeProtocolAsync(cancellationToken).ConfigureAwait(false); + AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager, runId, descriptor.Accepts, cancellationToken) .ConfigureAwait(false); - return await runHandle.WithCheckpointingAsync(() => new(new StreamingRun(runHandle))) - .ConfigureAwait(false); + await runHandle.EnqueueMessageAsync(input, cancellationToken).ConfigureAwait(false); + + if (descriptor.IsChatProtocol() && input is not TurnToken) + { + await runHandle.EnqueueMessageAsync(new TurnToken(emitEvents: true), cancellationToken).ConfigureAwait(false); + } + + return runHandle; } /// @@ -129,21 +132,13 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull { - var runHandle = await this.GetRunHandleWithTurnTokenAsync(workflow: workflow, input: input, checkpointManager: null, runId: runId, cancellationToken).ConfigureAwait(false); - - Run run = new(runHandle); - await run.RunToNextHaltAsync(cancellationToken).ConfigureAwait(false); - return run; - } - - /// - public async ValueTask RunAsync( - Workflow workflow, - TInput input, - string? runId = null, - CancellationToken cancellationToken = default) where TInput : notnull - { - var runHandle = await this.GetRunHandleWithTurnTokenAsync(workflow: workflow, input: input, checkpointManager: null, runId: runId, cancellationToken).ConfigureAwait(false); + AsyncRunHandle runHandle = await this.BeginRunHandlingChatProtocolAsync( + workflow, + input, + checkpointManager: null, + runId, + cancellationToken) + .ConfigureAwait(false); Run run = new(runHandle); await run.RunToNextHaltAsync(cancellationToken).ConfigureAwait(false); @@ -158,23 +153,13 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull { - var runHandle = await this.GetRunHandleWithTurnTokenAsync(workflow: workflow, input: input, checkpointManager: checkpointManager, runId: runId, cancellationToken).ConfigureAwait(false); - - Run run = new(runHandle); - await run.RunToNextHaltAsync(cancellationToken).ConfigureAwait(false); - return await runHandle.WithCheckpointingAsync(() => new ValueTask(run)) - .ConfigureAwait(false); - } - - /// - public async ValueTask> RunAsync( - Workflow workflow, - TInput input, - CheckpointManager checkpointManager, - string? runId = null, - CancellationToken cancellationToken = default) where TInput : notnull - { - var runHandle = await this.GetRunHandleWithTurnTokenAsync(workflow: workflow, input: input, checkpointManager: checkpointManager, runId: runId, cancellationToken).ConfigureAwait(false); + AsyncRunHandle runHandle = await this.BeginRunHandlingChatProtocolAsync( + workflow, + input, + checkpointManager, + runId, + cancellationToken) + .ConfigureAwait(false); Run run = new(runHandle); await run.RunToNextHaltAsync(cancellationToken).ConfigureAwait(false); @@ -196,63 +181,4 @@ public sealed class InProcessExecutionEnvironment : IWorkflowExecutionEnvironmen return await runHandle.WithCheckpointingAsync(() => new(new Run(runHandle))) .ConfigureAwait(false); } - - /// - public async ValueTask> ResumeAsync( - Workflow workflow, - CheckpointInfo fromCheckpoint, - CheckpointManager checkpointManager, - string? runId = null, - CancellationToken cancellationToken = default) where TInput : notnull - { - AsyncRunHandle runHandle = await this.ResumeRunAsync(workflow, checkpointManager, runId: runId, fromCheckpoint, [typeof(TInput)], cancellationToken) - .ConfigureAwait(false); - - return await runHandle.WithCheckpointingAsync(() => new(new Run(runHandle))) - .ConfigureAwait(false); - } - - // Helper to construct a RunHandle with the provided input enqueued. If the starting executor supports it, a TurnToken will be enqueued also. - private async ValueTask GetRunHandleWithTurnTokenAsync( - Workflow workflow, - TInput input, - CheckpointManager? checkpointManager, - string? runId, - CancellationToken cancellationToken) - { - var knownTypes = new List() { typeof(TInput) }; - var needsTurnToken = await StartingExecutorHandlesTurnTokenAsync(workflow).ConfigureAwait(false); - if (needsTurnToken) - { - knownTypes.Add(typeof(TurnToken)); - } - - AsyncRunHandle runHandle = await this.BeginRunAsync(workflow, checkpointManager: checkpointManager, runId: runId, knownTypes, cancellationToken) - .ConfigureAwait(false); - - await runHandle.EnqueueMessageAsync(input, cancellationToken).ConfigureAwait(false); - - if (needsTurnToken) - { - await runHandle.EnqueueMessageAsync(new TurnToken(emitEvents: true), cancellationToken).ConfigureAwait(false); - } - - return runHandle; - } - - /// - /// Helper method to detect if the starting executor of a given workflow accepts the provided input type as well as a TurnToken. - /// - private static async ValueTask StartingExecutorHandlesTurnTokenAsync(Workflow workflow) - { - if (workflow.Registrations.TryGetValue(workflow.StartExecutorId, out var registration)) - { - // Create instance to check type - Executor startExecutor = await registration.CreateInstanceAsync(string.Empty) - .ConfigureAwait(false); - return startExecutor.CanHandle(typeof(TInput)) && startExecutor.CanHandle(typeof(TurnToken)); - } - - return false; - } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProcessExecution.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProcessExecution.cs index 7f736e58f3..fe88f7ffff 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProcessExecution.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProcessExecution.cs @@ -41,51 +41,35 @@ public static class InProcessExecution /// internal static InProcessExecutionEnvironment Subworkflow { get; } = new(ExecutionMode.Subworkflow); + /// + public static ValueTask StreamAsync(Workflow workflow, string? runId = null, CancellationToken cancellationToken = default) + => Default.StreamAsync(workflow, runId, cancellationToken); + /// public static ValueTask StreamAsync(Workflow workflow, TInput input, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull => Default.StreamAsync(workflow, input, runId, cancellationToken); - /// - public static ValueTask StreamAsync(Workflow workflow, TInput input, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull - => Default.StreamAsync(workflow, input, runId, cancellationToken); + /// + public static ValueTask> StreamAsync(Workflow workflow, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) + => Default.StreamAsync(workflow, checkpointManager, runId, cancellationToken); /// public static ValueTask> StreamAsync(Workflow workflow, TInput input, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull => Default.StreamAsync(workflow, input, checkpointManager, runId, cancellationToken); - /// - public static ValueTask> StreamAsync(Workflow workflow, TInput input, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull - => Default.StreamAsync(workflow, input, checkpointManager, runId, cancellationToken); - /// public static ValueTask> ResumeStreamAsync(Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) => Default.ResumeStreamAsync(workflow, fromCheckpoint, checkpointManager, runId, cancellationToken); - /// - public static ValueTask> ResumeStreamAsync(Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull - => Default.ResumeStreamAsync(workflow, fromCheckpoint, checkpointManager, runId, cancellationToken); - /// public static ValueTask RunAsync(Workflow workflow, TInput input, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull => Default.RunAsync(workflow, input, runId, cancellationToken); - /// - public static ValueTask RunAsync(Workflow workflow, TInput input, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull - => Default.RunAsync(workflow, input, runId, cancellationToken); - /// public static ValueTask> RunAsync(Workflow workflow, TInput input, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull => Default.RunAsync(workflow, input, checkpointManager, runId, cancellationToken); - /// - public static ValueTask> RunAsync(Workflow workflow, TInput input, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull - => Default.RunAsync(workflow, input, checkpointManager, runId, cancellationToken); - /// public static ValueTask> ResumeAsync(Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) => Default.ResumeAsync(workflow, fromCheckpoint, checkpointManager, runId, cancellationToken); - - /// - public static ValueTask> ResumeAsync(Workflow workflow, CheckpointInfo fromCheckpoint, CheckpointManager checkpointManager, string? runId = null, CancellationToken cancellationToken = default) where TInput : notnull - => Default.ResumeAsync(workflow, fromCheckpoint, checkpointManager, runId, cancellationToken); } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/ProtocolDescriptor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/ProtocolDescriptor.cs new file mode 100644 index 0000000000..91adc4dbae --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/ProtocolDescriptor.cs @@ -0,0 +1,23 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Linq; + +namespace Microsoft.Agents.AI.Workflows; + +/// +/// Describes the protocol for communication with a or . +/// +public class ProtocolDescriptor +{ + /// + /// Get the collection of types accepted by the or . + /// + public IEnumerable Accepts { get; } + + internal ProtocolDescriptor(IEnumerable acceptedTypes) + { + this.Accepts = acceptedTypes.ToArray(); + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs index 8205d40b20..d2579871fa 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs @@ -85,42 +85,6 @@ public class Workflow this.Description = description; } - /// - /// Attempts to promote the current workflow to a type pre-checked instance that can handle input of type . - /// - /// The desired input type. - /// A type-parametrized workflow definitely able to process input of type or - /// if the workflow does not accept that type of input. - /// - internal async ValueTask?> TryPromoteAsync() - { - // Grab the start node, and make sure it has the right type? - if (!this.Registrations.TryGetValue(this.StartExecutorId, out ExecutorRegistration? startRegistration)) - { - // TODO: This should never be able to be hit - throw new InvalidOperationException($"Start executor with ID '{this.StartExecutorId}' is not bound."); - } - - // TODO: Can we cache this somehow to avoid having to instantiate a new one when running? - // Does that break some user expectations? - Executor startExecutor = await startRegistration.CreateInstanceAsync(string.Empty).ConfigureAwait(false); - - if (!startExecutor.InputTypes.Any(t => t.IsAssignableFrom(typeof(TInput)))) - { - // We have no handlers for the input type T, which means the built workflow will not be able to - // process messages of the desired type - return null; - } - - return new Workflow(this.StartExecutorId) - { - Registrations = this.Registrations, - Edges = this.Edges, - Ports = this.Ports, - OutputExecutors = this.OutputExecutors - }; - } - private bool _needsReset; private bool IsResettable => this.Registrations.Values.All(registration => !registration.IsUnresettableSharedInstance); @@ -215,27 +179,18 @@ public class Workflow await this.TryResetExecutorRegistrationsAsync().ConfigureAwait(false); } -} -/// -/// Represents a workflow that operates on data of type . -/// -/// The type of input to the workflow. -public class Workflow : Workflow -{ /// - /// Initializes a new instance of the class with the specified starting executor identifier + /// Retrieves a defining how to interact with this workflow. /// - /// The unique identifier of the starting executor for the workflow. Cannot be null. - /// Optional human-readable name for the workflow. - /// Optional description of what the workflow does. - public Workflow(string startExecutorId, string? name = null, string? description = null) - : base(startExecutorId, name, description) + /// The to monitor for cancellation requests. The default is . + /// A that represents that asynchronous operation. The result contains + /// a the protocol this follows. + public async ValueTask DescribeProtocolAsync(CancellationToken cancellationToken = default) { + ExecutorRegistration startExecutorRegistration = this.Registrations[this.StartExecutorId]; + Executor startExecutor = await startExecutorRegistration.CreateInstanceAsync(string.Empty) + .ConfigureAwait(false); + return startExecutor.DescribeProtocol(); } - - /// - /// Gets the type of input expected by the starting executor of the workflow. - /// - public Type InputType => typeof(T); } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowBuilder.cs index 94e4f20594..9797843fa7 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowBuilder.cs @@ -6,7 +6,6 @@ using System.Diagnostics; using System.Linq; using System.Text.Json; using System.Threading; -using System.Threading.Tasks; using Microsoft.Agents.AI.Workflows.Checkpointing; using Microsoft.Agents.AI.Workflows.Observability; using Microsoft.Shared.Diagnostics; @@ -441,34 +440,4 @@ public class WorkflowBuilder return workflow; } - - /// - /// Attempts to build a workflow instance configured to process messages of the specified input type. - /// - /// The desired input type for the workflow. - /// Thrown if the built workflow cannot process messages of the specified input type, - public async ValueTask> BuildAsync() where TInput : notnull - { - using Activity? activity = s_activitySource.StartActivity(ActivityNames.WorkflowBuild); - - Workflow? maybeWorkflow = await this.BuildInternal(activity) - .TryPromoteAsync() - .ConfigureAwait(false); - - if (maybeWorkflow is null) - { - var exception = new InvalidOperationException( - $"The built workflow cannot process input of type '{typeof(TInput).FullName}'."); - activity?.AddEvent(new ActivityEvent(EventNames.BuildError, tags: new() { - { Tags.BuildErrorMessage, exception.Message }, - { Tags.BuildErrorType, exception.GetType().FullName } - })); - activity?.CaptureException(exception); - throw exception; - } - - activity?.AddEvent(new ActivityEvent(EventNames.BuildCompleted)); - - return maybeWorkflow; - } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostAgent.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostAgent.cs index 54b36a8c64..98dc5903bf 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostAgent.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostAgent.cs @@ -18,12 +18,14 @@ internal sealed class WorkflowHostAgent : AIAgent private readonly string? _id; private readonly CheckpointManager? _checkpointManager; private readonly IWorkflowExecutionEnvironment _executionEnvironment; + private readonly Task _describeTask; private readonly ConcurrentDictionary _assignedRunIds = []; - public WorkflowHostAgent(Workflow> workflow, string? id = null, string? name = null, string? description = null, CheckpointManager? checkpointManager = null, IWorkflowExecutionEnvironment? executionEnvironment = null) + public WorkflowHostAgent(Workflow workflow, string? id = null, string? name = null, string? description = null, CheckpointManager? checkpointManager = null, IWorkflowExecutionEnvironment? executionEnvironment = null) { this._workflow = Throw.IfNull(workflow); + this._executionEnvironment = executionEnvironment ?? (workflow.AllowConcurrent ? InProcessExecution.Concurrent : InProcessExecution.OffThread); @@ -32,6 +34,9 @@ internal sealed class WorkflowHostAgent : AIAgent this._id = id; this.Name = name; this.Description = description; + + // Kick off the typecheck right away by starting the DescribeProtocol task. + this._describeTask = this._workflow.DescribeProtocolAsync().AsTask(); } public override string Id => this._id ?? base.Id; @@ -50,6 +55,12 @@ internal sealed class WorkflowHostAgent : AIAgent return result; } + private async ValueTask ValidateWorkflowAsync() + { + ProtocolDescriptor protocol = await this._describeTask.ConfigureAwait(false); + protocol.ThrowIfNotChatProtocol(); + } + public override AgentThread GetNewThread() => new WorkflowThread(this._workflow, this.GenerateNewId(), this._executionEnvironment, this._checkpointManager); public override AgentThread DeserializeThread(JsonElement serializedThread, JsonSerializerOptions? jsonSerializerOptions = null) @@ -75,6 +86,8 @@ internal sealed class WorkflowHostAgent : AIAgent AgentRunOptions? options = null, CancellationToken cancellationToken = default) { + await this.ValidateWorkflowAsync().ConfigureAwait(false); + WorkflowThread workflowThread = await this.UpdateThreadAsync(messages, thread, cancellationToken).ConfigureAwait(false); MessageMerger merger = new(); @@ -95,6 +108,8 @@ internal sealed class WorkflowHostAgent : AIAgent AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) { + await this.ValidateWorkflowAsync().ConfigureAwait(false); + WorkflowThread workflowThread = await this.UpdateThreadAsync(messages, thread, cancellationToken).ConfigureAwait(false); await foreach (AgentRunResponseUpdate update in workflowThread.InvokeStageAsync(cancellationToken) .ConfigureAwait(false) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostingExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostingExtensions.cs index e3a357a134..d48e99bf6e 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostingExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostingExtensions.cs @@ -1,8 +1,6 @@ // Copyright (c) Microsoft. All rights reserved. -using System; using System.Collections.Generic; -using System.Threading.Tasks; using Microsoft.Extensions.AI; namespace Microsoft.Agents.AI.Workflows; @@ -25,29 +23,6 @@ public static class WorkflowHostingExtensions /// for the in-process environments. /// public static AIAgent AsAgent( - this Workflow> workflow, - string? id = null, - string? name = null, - string? description = null, - CheckpointManager? checkpointManager = null, - IWorkflowExecutionEnvironment? executionEnvironment = null) - { - return new WorkflowHostAgent(workflow, id, name, description, checkpointManager, executionEnvironment); - } - - /// - /// Convert a workflow with the appropriate primary input type to an . - /// - /// The workflow to be hosted by the resulting - /// A unique id for the hosting . - /// A name for the hosting . - /// /// A description for the hosting . - /// A to enable persistence of run state. - /// Specify the execution environment to use when running the workflows. See - /// , and - /// for the in-process environments. - /// - public static async ValueTask AsAgentAsync( this Workflow workflow, string? id = null, string? name = null, @@ -55,15 +30,7 @@ public static class WorkflowHostingExtensions CheckpointManager? checkpointManager = null, IWorkflowExecutionEnvironment? executionEnvironment = null) { - Workflow>? maybeTyped = await workflow.TryPromoteAsync>() - .ConfigureAwait(false); - - if (maybeTyped is null) - { - throw new InvalidOperationException("Cannot host a workflow that does not accept List as an input"); - } - - return maybeTyped.AsAgent(id, name, description, checkpointManager, executionEnvironment); + return new WorkflowHostAgent(workflow, id, name, description, checkpointManager, executionEnvironment); } internal static FunctionCallContent ToFunctionCall(this ExternalRequest request) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowThread.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowThread.cs index 160785a49a..7866df5ec6 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowThread.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowThread.cs @@ -102,6 +102,8 @@ internal sealed class WorkflowThread : AgentThread private async ValueTask> CreateOrResumeRunAsync(List messages, CancellationToken cancellationToken = default) { + // The workflow is validated to be a ChatProtocol workflow by the WorkflowHostAgent before creating the thread, + // and does not need to be checked again here. if (this.LastCheckpoint is not null) { Checkpointed checkpointed = diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ChatProtocolExecutorTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ChatProtocolExecutorTests.cs index a689baee51..49987bc3ec 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ChatProtocolExecutorTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ChatProtocolExecutorTests.cs @@ -40,12 +40,23 @@ public class ChatProtocolExecutorTests } } + [Fact] + public void ChatProtocolExecutor_DescribedProtocol_IsChatProtocol() + { + // Arrange + TestChatProtocolExecutor executor = new(); + ProtocolDescriptor protocol = executor.DescribeProtocol(); + + // Act & Assert + protocol.Should().Match(protocol => protocol.IsChatProtocol()); + } + [Fact] public async Task ChatProtocolExecutor_Handles_ListOfChatMessagesAsync() { // Arrange - var executor = new TestChatProtocolExecutor(); - var context = new TestWorkflowContext(executor.Id); + TestChatProtocolExecutor executor = new(); + TestWorkflowContext context = new(executor.Id); List messages = [ @@ -68,8 +79,8 @@ public class ChatProtocolExecutorTests public async Task ChatProtocolExecutor_Handles_ArrayOfChatMessagesAsync() { // Arrange - var executor = new TestChatProtocolExecutor(); - var context = new TestWorkflowContext(executor.Id); + TestChatProtocolExecutor executor = new(); + TestWorkflowContext context = new(executor.Id); ChatMessage[] messages = [ @@ -94,8 +105,8 @@ public class ChatProtocolExecutorTests public async Task ChatProtocolExecutor_Handles_SingleChatMessageAsync() { // Arrange - var executor = new TestChatProtocolExecutor(); - var context = new TestWorkflowContext(executor.Id); + TestChatProtocolExecutor executor = new(); + TestWorkflowContext context = new(executor.Id); var message = new ChatMessage(ChatRole.User, "Single message"); @@ -112,8 +123,8 @@ public class ChatProtocolExecutorTests [Fact] public async Task ChatProtocolExecutor_AccumulatesAndClearsMessagesPerTurnAsync() { - var executor = new TestChatProtocolExecutor(); - var context = new TestWorkflowContext(executor.Id); + TestChatProtocolExecutor executor = new(); + TestWorkflowContext context = new(executor.Id); // Send multiple message batches before taking a turn await executor.ExecuteAsync(new ChatMessage(ChatRole.User, "Message 1"), new TypeId(typeof(ChatMessage)), context); @@ -147,12 +158,12 @@ public class ChatProtocolExecutorTests [Fact] public async Task ChatProtocolExecutor_WithStringRole_ConvertsStringToMessageAsync() { - var executor = new TestChatProtocolExecutor( + TestChatProtocolExecutor executor = new( options: new ChatProtocolExecutorOptions { StringMessageChatRole = ChatRole.User }); - var context = new TestWorkflowContext(executor.Id); + TestWorkflowContext context = new(executor.Id); await executor.ExecuteAsync("String message", new TypeId(typeof(string)), context); await executor.TakeTurnAsync(new TurnToken(emitEvents: false), context); @@ -165,8 +176,8 @@ public class ChatProtocolExecutorTests [Fact] public async Task ChatProtocolExecutor_EmptyCollection_HandledCorrectlyAsync() { - var executor = new TestChatProtocolExecutor(); - var context = new TestWorkflowContext(executor.Id); + TestChatProtocolExecutor executor = new(); + TestWorkflowContext context = new(executor.Id); await executor.ExecuteAsync(new List(), new TypeId(typeof(List)), context); await executor.ExecuteAsync(Array.Empty(), new TypeId(typeof(ChatMessage[])), context); @@ -181,8 +192,8 @@ public class ChatProtocolExecutorTests [InlineData(typeof(ChatMessage[]))] public async Task ChatProtocolExecutor_RoutesCollectionTypesAsync(Type collectionType) { - var executor = new TestChatProtocolExecutor(); - var context = new TestWorkflowContext(executor.Id); + TestChatProtocolExecutor executor = new(); + TestWorkflowContext context = new(executor.Id); var sourceMessages = new[] { new ChatMessage(ChatRole.User, "Test message") }; object messagesToSend = collectionType == typeof(List) ? sourceMessages.ToList() : sourceMessages; @@ -197,8 +208,8 @@ public class ChatProtocolExecutorTests [Fact] public async Task ChatProtocolExecutor_MultipleTurns_EachTurnProcessesSeparatelyAsync() { - var executor = new TestChatProtocolExecutor(); - var context = new TestWorkflowContext(executor.Id); + TestChatProtocolExecutor executor = new(); + TestWorkflowContext context = new(executor.Id); await executor.ExecuteAsync(new List { new(ChatRole.User, "Turn 1") }, new TypeId(typeof(List)), context); await executor.TakeTurnAsync(new TurnToken(emitEvents: false), context); @@ -217,8 +228,8 @@ public class ChatProtocolExecutorTests [Fact] public async Task ChatProtocolExecutor_InitialWorkflowMessages_RoutedCorrectlyAsync() { - var executor = new TestChatProtocolExecutor(); - var context = new TestWorkflowContext(executor.Id); + TestChatProtocolExecutor executor = new(); + TestWorkflowContext context = new(executor.Id); List initialMessages = [new ChatMessage(ChatRole.User, "Kick off the workflow")]; diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/JsonSerializationTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/JsonSerializationTests.cs index cd0f910ddb..7afbe96abe 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/JsonSerializationTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/JsonSerializationTests.cs @@ -155,7 +155,7 @@ public class JsonSerializationTests private static RequestPortInfo IntToString => RequestPort.Create(IntToStringId).ToPortInfo(); private static RequestPortInfo StringToInt => RequestPort.Create(StringToIntId).ToPortInfo(); - private static ValueTask> CreateTestWorkflowAsync() + private static Workflow CreateTestWorkflow() { ForwardMessageExecutor forwardString = new(ForwardStringId); ForwardMessageExecutor forwardInt = new(ForwardIntId); @@ -169,12 +169,12 @@ public class JsonSerializationTests .AddEdge(forwardInt, intToString) .AddEdge(intToString, StreamingAggregators.Last().AsExecutor("Aggregate")); - return builder.BuildAsync(); + return builder.Build(); } - internal static async ValueTask CreateTestWorkflowInfoAsync() + internal static WorkflowInfo CreateTestWorkflowInfo() { - Workflow testWorkflow = await CreateTestWorkflowAsync().ConfigureAwait(false); + Workflow testWorkflow = CreateTestWorkflow(); return testWorkflow.ToWorkflowInfo(); } @@ -232,7 +232,7 @@ public class JsonSerializationTests [Fact] public async Task Test_WorkflowInfo_JsonRoundtripAsync() { - WorkflowInfo prototype = await CreateTestWorkflowInfoAsync(); + WorkflowInfo prototype = CreateTestWorkflowInfo(); JsonMarshaller marshaller = new(); @@ -637,7 +637,7 @@ public class JsonSerializationTests [Fact] public async Task Test_Checkpoint_JsonRoundTripAsync() { - WorkflowInfo testWorkflowInfo = await CreateTestWorkflowInfoAsync(); + WorkflowInfo testWorkflowInfo = CreateTestWorkflowInfo(); Checkpoint prototype = new(12, testWorkflowInfo, TestRunnerStateData, TestStateData, TestEdgeState, TestParentCheckpointInfo); Checkpoint result = RunJsonRoundtrip(prototype, TestCustomSerializedJsonOptions); diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/RepresentationTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/RepresentationTests.cs index 85f5baee6a..fa5ef22903 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/RepresentationTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/RepresentationTests.cs @@ -157,23 +157,17 @@ public class RepresentationTests [Fact] public async Task Test_Sample_WorkflowInfosAsync() { - Workflow workflowStep1 = (await Step1EntryPoint.WorkflowInstance.TryPromoteAsync())!; - RunWorkflowInfoMatchTest(workflowStep1); - - Workflow workflowStep2 = (await Step2EntryPoint.WorkflowInstance.TryPromoteAsync())!; - RunWorkflowInfoMatchTest(workflowStep2); - - RunWorkflowInfoMatchTest((await Step3EntryPoint.WorkflowInstance.TryPromoteAsync())!); - - RunWorkflowInfoMatchTest((await Step4EntryPoint.WorkflowInstance.TryPromoteAsync())!); - + RunWorkflowInfoMatchTest(Step1EntryPoint.WorkflowInstance); + RunWorkflowInfoMatchTest(Step2EntryPoint.WorkflowInstance); + RunWorkflowInfoMatchTest(Step3EntryPoint.WorkflowInstance); + RunWorkflowInfoMatchTest(Step4EntryPoint.WorkflowInstance); // Step 5 reuses the workflow from Step 4, so we don't need to test it separately. - RunWorkflowInfoMatchTest((await Step6EntryPoint.CreateWorkflow(2).TryPromoteAsync>())!); + RunWorkflowInfoMatchTest(Step6EntryPoint.CreateWorkflow(maxTurns: 2)); // Step 7 reuses the workflow from Step 6, so we don't need to test it separately. - RunWorkflowInfoMatchTest(workflowStep1, workflowStep2, expect: false); + RunWorkflowInfoMatchTest(Step1EntryPoint.WorkflowInstance, Step2EntryPoint.WorkflowInstance, expect: false); - static void RunWorkflowInfoMatchTest(Workflow workflow, Workflow? comparator = null, bool expect = true) + static void RunWorkflowInfoMatchTest(Workflow workflow, Workflow? comparator = null, bool expect = true) { comparator ??= workflow; diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/01_Simple_Workflow_Sequential.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/01_Simple_Workflow_Sequential.cs index 72dfca59e4..c6d33e13d7 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/01_Simple_Workflow_Sequential.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/01_Simple_Workflow_Sequential.cs @@ -27,7 +27,7 @@ internal static class Step1EntryPoint public static async ValueTask RunAsync(TextWriter writer, IWorkflowExecutionEnvironment environment) { - StreamingRun run = await environment.StreamAsync(WorkflowInstance, "Hello, World!").ConfigureAwait(false); + StreamingRun run = await environment.StreamAsync(WorkflowInstance, input: "Hello, World!").ConfigureAwait(false); await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false)) { diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/02_Simple_Workflow_Condition.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/02_Simple_Workflow_Condition.cs index 50e65dc55c..9ee50ae3fb 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/02_Simple_Workflow_Condition.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/02_Simple_Workflow_Condition.cs @@ -31,7 +31,7 @@ internal static class Step2EntryPoint public static async ValueTask RunAsync(TextWriter writer, IWorkflowExecutionEnvironment environment, string input = "This is a spam message.") { - StreamingRun handle = await environment.StreamAsync(WorkflowInstance, input).ConfigureAwait(false); + StreamingRun handle = await environment.StreamAsync(WorkflowInstance, input: input).ConfigureAwait(false); await foreach (WorkflowEvent evt in handle.WatchStreamAsync().ConfigureAwait(false)) { switch (evt) diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/04_Simple_Workflow_ExternalRequest.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/04_Simple_Workflow_ExternalRequest.cs index 69d21600c4..cc417e727a 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/04_Simple_Workflow_ExternalRequest.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/04_Simple_Workflow_ExternalRequest.cs @@ -23,12 +23,6 @@ internal static class Step4EntryPoint .Build(); } - public static ValueTask?> GetPromotedWorklowInstanceAsync() - { - Workflow workflow = CreateWorkflowInstance(out _); - return workflow.TryPromoteAsync(); - } - public static Workflow WorkflowInstance { get diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs index 9c74bc47e8..7b6d86afb1 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs @@ -1,6 +1,5 @@ // Copyright (c) Microsoft. All rights reserved. -using System.Collections.Generic; using System.IO; using System.Threading.Tasks; using Microsoft.Extensions.AI; @@ -11,9 +10,7 @@ internal static class Step7EntryPoint { public static async ValueTask RunAsync(TextWriter writer, int maxSteps = 2) { - Workflow> workflow = (await Step6EntryPoint.CreateWorkflow(maxSteps) - .TryPromoteAsync>() - .ConfigureAwait(false))!; + Workflow workflow = Step6EntryPoint.CreateWorkflow(maxSteps); AIAgent agent = workflow.AsAgent("group-chat-agent", "Group Chat Agent"); diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/StateManagerTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/StateManagerTests.cs index fc16fd6600..13c21025fa 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/StateManagerTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/StateManagerTests.cs @@ -538,7 +538,7 @@ public class StateManagerTests Dictionary exportedState = await manager.ExportStateAsync(); Dictionary serializedState = JsonSerializationTests.RunJsonRoundtrip(exportedState); - Checkpoint testCheckpoint = new(0, await JsonSerializationTests.CreateTestWorkflowInfoAsync(), new([], [], []), serializedState, new()); + Checkpoint testCheckpoint = new(0, JsonSerializationTests.CreateTestWorkflowInfo(), new([], [], []), serializedState, new()); manager = new(); await manager.ImportStateAsync(testCheckpoint);