diff --git a/dotnet/samples/GettingStarted/Workflows/Declarative/Program.cs b/dotnet/samples/GettingStarted/Workflows/Declarative/Program.cs index 60a87c6c5b..073650274c 100644 --- a/dotnet/samples/GettingStarted/Workflows/Declarative/Program.cs +++ b/dotnet/samples/GettingStarted/Workflows/Declarative/Program.cs @@ -62,7 +62,7 @@ internal sealed class Program // Run the workflow, just like any other workflow string input = this.GetWorkflowInput(); - CheckpointManager checkpointManager = CheckpointManager.Default; + CheckpointManager checkpointManager = CheckpointManager.CreateInMemory(); Checkpointed run = await InProcessExecution.StreamAsync(workflow, input, checkpointManager); bool isComplete = false; @@ -88,7 +88,7 @@ internal sealed class Program // Restore the latest checkpoint. Debug.WriteLine($"RESTORE #{this.LastCheckpoint.CheckpointId}"); Notify("\nWORKFLOW: Restore"); - run = await InProcessExecution.ResumeStreamAsync(workflow, this.LastCheckpoint, checkpointManager); + run = await InProcessExecution.ResumeStreamAsync(workflow, this.LastCheckpoint, checkpointManager, run.Run.RunId); } else { @@ -178,6 +178,7 @@ internal sealed class Program } else { + await run.Run.EndRunAsync().ConfigureAwait(false); return requestInfo.Request; } break; diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/AzureAgentProvider.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/AzureAgentProvider.cs index 40d3eb6ba4..7a0780e456 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/AzureAgentProvider.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/AzureAgentProvider.cs @@ -1,4 +1,5 @@ // Copyright (c) Microsoft. All rights reserved. + using System.Collections.Generic; using System.Linq; using System.Net.Http; diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/AgentProviderExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/AgentProviderExtensions.cs index 184913280f..e6c717791c 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/AgentProviderExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/AgentProviderExtensions.cs @@ -1,7 +1,7 @@ // Copyright (c) Microsoft. All rights reserved. using System.Collections.Generic; -using System.Runtime.CompilerServices; +using System.Linq; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.AI; @@ -10,7 +10,7 @@ namespace Microsoft.Agents.AI.Workflows.Declarative.Extensions; internal static class AgentProviderExtensions { - public static async IAsyncEnumerable InvokeAgentAsync( + public static async ValueTask InvokeAgentAsync( this WorkflowAgentProvider agentProvider, string executorId, IWorkflowContext context, @@ -19,35 +19,68 @@ internal static class AgentProviderExtensions bool autoSend, string? additionalInstructions = null, IEnumerable? inputMessages = null, - [EnumeratorCancellation] CancellationToken cancellationToken = default) + CancellationToken cancellationToken = default) { + // Get the specified agent. AIAgent agent = await agentProvider.GetAgentAsync(agentName, cancellationToken).ConfigureAwait(false); + // Prepare the run options. ChatClientAgentRunOptions options = new( new ChatOptions() { + ConversationId = conversationId, Instructions = additionalInstructions, }); - AgentThread agentThread = conversationId is not null && agent is ChatClientAgent chatClientAgent ? chatClientAgent.GetNewThread(conversationId) : agent.GetNewThread(); + // Initialize the agent thread. IAsyncEnumerable agentUpdates = inputMessages is not null ? - agent.RunStreamingAsync([.. inputMessages], agentThread, options, cancellationToken) : - agent.RunStreamingAsync(agentThread, options, cancellationToken); + agent.RunStreamingAsync([.. inputMessages], null, options, cancellationToken) : + agent.RunStreamingAsync(null, options, cancellationToken); + // Enable "autoSend" behavior if this is the workflow conversation. + bool isWorkflowConversation = context.IsWorkflowConversation(conversationId); + autoSend &= isWorkflowConversation; + + // Process the agent response updates. + List updates = []; await foreach (AgentRunResponseUpdate update in agentUpdates.ConfigureAwait(false)) { await AssignConversationIdAsync(((ChatResponseUpdate?)update.RawRepresentation)?.ConversationId).ConfigureAwait(false); + updates.Add(update); + if (autoSend) { await context.AddEventAsync(new AgentRunUpdateEvent(executorId, update)).ConfigureAwait(false); } - - yield return update; } + AgentRunResponse response = updates.ToAgentRunResponse(); + + if (autoSend) + { + await context.AddEventAsync(new AgentRunResponseEvent(executorId, response)).ConfigureAwait(false); + } + + if (autoSend && !isWorkflowConversation && conversationId is not null) + { + // Copy messages with content that aren't function calls or results. + IEnumerable messages = + response.Messages.Where( + message => + !string.IsNullOrEmpty(message.Text) && + !message.Contents.OfType().Any() && + !message.Contents.OfType().Any()); + foreach (ChatMessage message in messages) + { + await agentProvider.CreateMessageAsync(conversationId, message, cancellationToken).ConfigureAwait(false); + } + } + + return response; + async ValueTask AssignConversationIdAsync(string? assignValue) { if (assignValue is not null && conversationId is null) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/IWorkflowContextExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/IWorkflowContextExtensions.cs index 5afbb31f7c..3171e2ac2a 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/IWorkflowContextExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/IWorkflowContextExtensions.cs @@ -1,5 +1,6 @@ // Copyright (c) Microsoft. All rights reserved. +using System; using System.Threading; using System.Threading.Tasks; using Microsoft.Agents.AI.Workflows.Declarative.Interpreter; @@ -46,6 +47,17 @@ internal static class IWorkflowContextExtensions await context.AddEventAsync(new ConversationUpdateEvent(conversationId)).ConfigureAwait(false); } + public static bool IsWorkflowConversation(this IWorkflowContext context, string? conversationId) + { + if (string.IsNullOrWhiteSpace(conversationId)) + { + return false; + } + + StringValue workflowId = (StringValue)context.ReadState(SystemScope.Names.ConversationId, VariableScopeNames.System); + return workflowId.Value.Equals(conversationId, StringComparison.Ordinal); + } + private static DeclarativeWorkflowContext DeclarativeContext(IWorkflowContext context) { if (context is not DeclarativeWorkflowContext declarativeContext) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeAzureAgentExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeAzureAgentExecutor.cs index b1e98f54a4..757228b10a 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeAzureAgentExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeAzureAgentExecutor.cs @@ -1,7 +1,6 @@ // Copyright (c) Microsoft. All rights reserved. using System.Collections.Generic; -using System.Linq; using System.Threading; using System.Threading.Tasks; using Microsoft.Agents.AI.Workflows.Declarative.Extensions; @@ -29,12 +28,7 @@ internal sealed class InvokeAzureAgentExecutor(InvokeAzureAgent model, WorkflowA bool autoSend = this.GetAutoSendValue(); IEnumerable? inputMessages = this.GetInputMessages(); - AgentRunResponse agentResponse = agentProvider.InvokeAgentAsync(this.Id, context, agentName, conversationId, autoSend, additionalInstructions, inputMessages, cancellationToken).ToEnumerable().ToAgentRunResponse(); - - if (autoSend) - { - await context.AddEventAsync(new AgentRunResponseEvent(this.Id, agentResponse)).ConfigureAwait(false); - } + AgentRunResponse agentResponse = await agentProvider.InvokeAgentAsync(this.Id, context, agentName, conversationId, autoSend, additionalInstructions, inputMessages, cancellationToken).ConfigureAwait(false); await this.AssignAsync(this.AgentOutput?.Messages?.Path, agentResponse.Messages.ToTable(), context).ConfigureAwait(false);