diff --git a/dotnet/src/Microsoft.Agents.Workflows/AIAgentsAbstractionsExtensions.cs b/dotnet/src/Microsoft.Agents.Workflows/AIAgentsAbstractionsExtensions.cs new file mode 100644 index 0000000000..d627d01788 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.Workflows/AIAgentsAbstractionsExtensions.cs @@ -0,0 +1,42 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Collections.Generic; +using System.Diagnostics; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.AI.Agents; + +namespace Microsoft.Agents.Workflows; + +internal static class AIAgentsAbstractionsExtensions +{ + public static ChatMessage ToChatMessage(this AgentRunResponseUpdate update) + { + return new ChatMessage + { + AuthorName = update.AuthorName, + Contents = update.Contents, + Role = update.Role ?? ChatRole.User, + CreatedAt = update.CreatedAt, + MessageId = update.MessageId, + RawRepresentation = update.RawRepresentation, + }; + } + + public static ChatMessage UpdateWith(this ChatMessage baseMessage, AgentRunResponseUpdate update) + { + Debug.Assert(update.MessageId == null || baseMessage.MessageId == update.MessageId); + + List mergedContent = new(baseMessage.Contents); + mergedContent.AddRange(update.Contents); + + return new ChatMessage + { + AuthorName = update.AuthorName ?? baseMessage.AuthorName, + Contents = mergedContent, + Role = update.Role ?? baseMessage.Role, + CreatedAt = update.CreatedAt ?? baseMessage.CreatedAt, + MessageId = baseMessage.MessageId, + RawRepresentation = update.RawRepresentation ?? baseMessage.RawRepresentation, + }; + } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/AgentRunEvent.cs b/dotnet/src/Microsoft.Agents.Workflows/AgentRunEvent.cs deleted file mode 100644 index 7fc66990a9..0000000000 --- a/dotnet/src/Microsoft.Agents.Workflows/AgentRunEvent.cs +++ /dev/null @@ -1,26 +0,0 @@ -// Copyright (c) Microsoft. All rights reserved. - -using Microsoft.Extensions.AI.Agents; - -namespace Microsoft.Agents.Workflows; - -/// -/// Event triggered when an agent run is completed. -/// -public class AgentRunEvent : ExecutorEvent -{ - /// - /// Initializes a new instance of the class. - /// - /// The identifier of the executor that generated this event. - /// - public AgentRunEvent(string executorId, AgentRunResponse? response = null) : base(executorId, data: response) - { - this.Response = response; - } - - /// - /// Gets the content of the agent response. - /// - public AgentRunResponse? Response { get; } -} diff --git a/dotnet/src/Microsoft.Agents.Workflows/AgentRunUpdateEvent.cs b/dotnet/src/Microsoft.Agents.Workflows/AgentRunUpdateEvent.cs new file mode 100644 index 0000000000..0972fdd871 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.Workflows/AgentRunUpdateEvent.cs @@ -0,0 +1,37 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Collections.Generic; +using Microsoft.Extensions.AI.Agents; + +namespace Microsoft.Agents.Workflows; + +/// +/// Event triggered when an agent run produces an update. +/// +public class AgentRunUpdateEvent : ExecutorEvent +{ + /// + /// Initializes a new instance of the class. + /// + /// The identifier of the executor that generated this event. + /// + public AgentRunUpdateEvent(string executorId, AgentRunResponseUpdate update) : base(executorId, data: update) + { + this.Update = update; + } + + /// + /// Gets the content of the agent response. + /// + public AgentRunResponseUpdate Update { get; } + + /// + /// Converts this event to an containing just this update. + /// + /// + public AgentRunResponse AsResponse() + { + IEnumerable updates = [this.Update]; + return updates.ToAgentRunResponse(); + } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/MessageMerger.cs b/dotnet/src/Microsoft.Agents.Workflows/MessageMerger.cs new file mode 100644 index 0000000000..021330a90d --- /dev/null +++ b/dotnet/src/Microsoft.Agents.Workflows/MessageMerger.cs @@ -0,0 +1,217 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.Linq; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.AI.Agents; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Agents.Workflows; + +internal class MessageMerger +{ + private class ResponseMergeState(string? responseId) + { + public string? ResponseId { get; } = responseId; + + public Dictionary> UpdatesByMessageId { get; } = new(); + public List DanglingUpdates { get; } = new(); + + public void AddUpdate(AgentRunResponseUpdate update) + { + if (update.MessageId is null) + { + this.DanglingUpdates.Add(update); + } + else + { + if (!this.UpdatesByMessageId.TryGetValue(update.MessageId, out List? updates)) + { + this.UpdatesByMessageId[update.MessageId] = updates = new List(); + } + + updates.Add(update); + } + } + + private AgentRunResponse ComputeResponse(List updates) => updates.ToAgentRunResponse(); + + public AgentRunResponse ComputeMerged(string messageId) + { + if (this.UpdatesByMessageId.TryGetValue(Throw.IfNull(messageId), out List? updates)) + { + return updates.ToAgentRunResponse(); + } + + throw new KeyNotFoundException($"No updates found for message ID '{messageId}' in response '{this.ResponseId}'."); + } + + public AgentRunResponse ComputeDangling() + { + if (this.DanglingUpdates.Count == 0) + { + throw new InvalidOperationException("No dangling updates to compute a response from."); + } + + return this.DanglingUpdates.ToAgentRunResponse(); + } + + public List ComputeFlattened() + { + List result = this.UpdatesByMessageId.Keys.Select(AggregateUpdatesToMessage) + .ToList(); + + result.AddRange(this.ComputeDangling().Messages); + return result; + + ChatMessage AggregateUpdatesToMessage(string messageId) + { + List updates = this.UpdatesByMessageId[messageId]; + if (updates.Count == 0) + { + throw new InvalidOperationException($"No updates found for message ID '{messageId}' in response '{this.ResponseId}'."); + } + + return updates.Aggregate(null, + (ChatMessage? previous, AgentRunResponseUpdate current) => + { + return previous == null + ? current.ToChatMessage() + : previous.UpdateWith(current); + })!; + } + } + } + + private readonly Dictionary _mergeStates = new(); + private readonly ResponseMergeState _danglingState = new(null); + + public void AddUpdate(AgentRunResponseUpdate update) + { + if (update.ResponseId is null) + { + this._danglingState.DanglingUpdates.Add(update); + } + else + { + if (!this._mergeStates.TryGetValue(update.ResponseId, out ResponseMergeState? state)) + { + this._mergeStates[update.ResponseId] = state = new ResponseMergeState(update.ResponseId); + } + + state.AddUpdate(update); + } + } + + private int CompareByDateTimeOffset(AgentRunResponse left, AgentRunResponse right) + { + const int LESS = -1, EQ = 0, GREATER = 1; + + if (left.CreatedAt == right.CreatedAt) + { + return EQ; + } + + if (!left.CreatedAt.HasValue) + { + return GREATER; + } + + if (!right.CreatedAt.HasValue) + { + return LESS; + } + + return left.CreatedAt.Value.CompareTo(right.CreatedAt.Value); + } + + public AgentRunResponse ComputeMerged(string primaryResponseId) + { + List messages = []; + Dictionary responses = new(); + + foreach (string responseId in this._mergeStates.Keys) + { + ResponseMergeState mergeState = this._mergeStates[responseId]; + + List responseList = mergeState.UpdatesByMessageId.Keys.Select(messageId => mergeState.ComputeMerged(messageId)).ToList(); + if (mergeState.DanglingUpdates.Count > 0) + { + responseList.Add(mergeState.ComputeDangling()); + } + + responseList.Sort(this.CompareByDateTimeOffset); + responses[responseId] = responseList.Aggregate(MergeResponses); + messages.AddRange(responses[responseId].Messages); + } + + messages.AddRange(this._danglingState.ComputeFlattened()); + return new AgentRunResponse(messages) + { + ResponseId = primaryResponseId, + }; + + AgentRunResponse MergeResponses(AgentRunResponse? current, AgentRunResponse incoming) + { + if (current is null) + { + return incoming; + } + + if (current.ResponseId != incoming.ResponseId) + { + throw new InvalidOperationException($"Cannot merge responses with different IDs: '{current.ResponseId}' and '{incoming.ResponseId}'."); + } + + List rawRepresentation = current.RawRepresentation as List ?? []; + rawRepresentation.Add(incoming.RawRepresentation); + + return new() + { + AgentId = incoming.AgentId ?? current.AgentId, + AdditionalProperties = incoming.AdditionalProperties ?? current.AdditionalProperties, + CreatedAt = incoming.CreatedAt ?? current.CreatedAt, + Messages = current.Messages.Concat(incoming.Messages).ToList(), + ResponseId = current.ResponseId, + RawRepresentation = rawRepresentation, + Usage = Merge(current.Usage, incoming.Usage), + }; + } + + static UsageDetails? Merge(UsageDetails? current, UsageDetails? incoming) + { + if (current == null) + { + return incoming; + } + + AdditionalPropertiesDictionary? additionalCounts = current.AdditionalCounts; + if (incoming == null) + { + return current; + } + + if (additionalCounts == null) + { + additionalCounts = incoming.AdditionalCounts; + } + else if (incoming.AdditionalCounts != null) + { + foreach (string key in incoming.AdditionalCounts.Keys) + { + additionalCounts[key] = incoming.AdditionalCounts[key] + + (additionalCounts.TryGetValue(key, out long? existingCount) ? existingCount.Value : 0); + } + } + + return new UsageDetails + { + InputTokenCount = current.InputTokenCount + incoming.InputTokenCount, + OutputTokenCount = current.OutputTokenCount + incoming.OutputTokenCount, + TotalTokenCount = current.TotalTokenCount + incoming.TotalTokenCount, + AdditionalCounts = additionalCounts, + }; + } + } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/Specialized/AIAgentHostExecutor.cs b/dotnet/src/Microsoft.Agents.Workflows/Specialized/AIAgentHostExecutor.cs index fbb97d8802..16575b1f95 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/Specialized/AIAgentHostExecutor.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/Specialized/AIAgentHostExecutor.cs @@ -1,7 +1,6 @@ // Copyright (c) Microsoft. All rights reserved. using System.Collections.Generic; -using System.Linq; using System.Threading.Tasks; using Microsoft.Extensions.AI; using Microsoft.Extensions.AI.Agents; @@ -10,13 +9,15 @@ namespace Microsoft.Agents.Workflows.Specialized; internal class AIAgentHostExecutor : Executor { + private readonly bool _emitEvents; private readonly AIAgent _agent; private readonly List _pendingMessages = new(); private AgentThread? _thread = null; - public AIAgentHostExecutor(AIAgent agent) : base(id: agent.Id) + public AIAgentHostExecutor(AIAgent agent, bool emitEvents = false) : base(id: agent.Id) { this._agent = agent; + this._emitEvents = emitEvents; } private AgentThread EnsureThread() @@ -50,17 +51,35 @@ internal class AIAgentHostExecutor : Executor public async ValueTask TakeTurnAsync(TurnToken token, IWorkflowContext context) { - // TODO: Ideally we want to be able to split the Run across multiple super-steps so that we can stream out - // incremental updates from the chat model. - AgentRunResponse runResponse = await this._agent.RunAsync(this._pendingMessages, this.EnsureThread()) - .ConfigureAwait(false); + bool emitEvents = token.EmitEvents.HasValue ? token.EmitEvents.Value : this._emitEvents; + IAsyncEnumerable agentStream = this._agent.RunStreamingAsync(this._pendingMessages, this.EnsureThread()); - if (token.EmitEvents) + List updates = new(); + await foreach (AgentRunResponseUpdate update in agentStream.ConfigureAwait(false)) { - await context.AddEventAsync(new AgentRunEvent(this.Id, runResponse)).ConfigureAwait(false); + if (emitEvents) + { + await context.AddEventAsync(new AgentRunUpdateEvent(this.Id, update)).ConfigureAwait(false); + } + + // TODO: FunctionCall request handling, and user info request handling. + // In some sense: We should just let it be handled as a ChatMessage, though we should consider + // providing some mechanisms to help the user complete the request, or route it out of the + // workflow. + + updates.Add(update); + ChatMessage message = new(update.Role ?? ChatRole.Assistant, update.Contents) + { + AuthorName = update.AuthorName, + CreatedAt = update.CreatedAt, + MessageId = update.MessageId, + RawRepresentation = update.RawRepresentation, + AdditionalProperties = update.AdditionalProperties + }; + + await context.SendMessageAsync(message).ConfigureAwait(false); } - await context.SendMessageAsync(runResponse.Messages.ToList()).ConfigureAwait(false); await context.SendMessageAsync(token).ConfigureAwait(false); } } diff --git a/dotnet/src/Microsoft.Agents.Workflows/TurnToken.cs b/dotnet/src/Microsoft.Agents.Workflows/TurnToken.cs index a561607799..f6448b64a1 100644 --- a/dotnet/src/Microsoft.Agents.Workflows/TurnToken.cs +++ b/dotnet/src/Microsoft.Agents.Workflows/TurnToken.cs @@ -9,10 +9,11 @@ namespace Microsoft.Agents.Workflows; /// a response to accumulated . /// /// Whether to raise AgentRunEvents for this executor. -public class TurnToken(bool emitEvents = false) +public class TurnToken(bool? emitEvents = null) { /// - /// Gets a value indicating whether events are emitted by the receiving executor. + /// Gets a value indicating whether events are emitted by the receiving executor. If the + /// value is not set, defaults to the configuration in the executor. /// - public bool EmitEvents => emitEvents; + public bool? EmitEvents => emitEvents; } diff --git a/dotnet/src/Microsoft.Agents.Workflows/WorkflowHostAgent.cs b/dotnet/src/Microsoft.Agents.Workflows/WorkflowHostAgent.cs new file mode 100644 index 0000000000..98b7a05fda --- /dev/null +++ b/dotnet/src/Microsoft.Agents.Workflows/WorkflowHostAgent.cs @@ -0,0 +1,150 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.AI.Agents; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Agents.Workflows; + +internal class WorkflowHostAgent : AIAgent +{ + private readonly Workflow> _workflow; + private readonly string? _id, _name; + + private readonly ConcurrentDictionary _assignedRunIds = new(); + private readonly Dictionary _runningWorkflows = new(); + + public WorkflowHostAgent(Workflow> workflow, string? id = null, string? name = null) + { + this._workflow = Throw.IfNull(workflow, nameof(workflow)); + + this._id = id; + this._name = name; + } + + public override string? Name => this._name; + public override string Id => this._id ?? base.Id; + + private string GenerateNewId() + { + string result; + + do + { + result = Guid.NewGuid().ToString("N"); + } while (!this._assignedRunIds.TryAdd(result, result)); + + return result; + } + + public override AgentThread GetNewThread() => new WorkflowThread(this.Id, this.Name, this.GenerateNewId()); + + private async + IAsyncEnumerable InvokeStageAsync( + WorkflowThread conversation, + [EnumeratorCancellation] CancellationToken cancellation = default) + { + string runId = conversation.RunId; + List messages = conversation.MessageStore.GetFromBookmark().ToList(); + + try + { + // technically there is a race condition here between assigning the ID, and checking if it exists + // in the case of new threads. + if (!this._runningWorkflows.TryGetValue(runId, out StreamingRun? run)) + { + run = await InProcessExecution.StreamAsync>(this._workflow, messages, cancellation) + .ConfigureAwait(false); + this._runningWorkflows[runId] = run; + } + else + { + bool sentMessages = await run.TrySendMessageAsync(messages).ConfigureAwait(false); + Debug.Assert(sentMessages, "Hosted workflow is required to take List as input."); + } + + await run.TrySendMessageAsync(new TurnToken(emitEvents: true)).ConfigureAwait(false); + await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false, cancellation) + .ConfigureAwait(false) + .WithCancellation(cancellation)) + { + switch (evt) + { + case AgentRunUpdateEvent agentUpdate: + yield return agentUpdate.Update; + break; + case RequestInfoEvent requestInfo: + FunctionCallContent fcContent = requestInfo.Request.ToFunctionCall(); + AgentRunResponseUpdate update = conversation.CreateUpdate(fcContent); + yield return update; + break; + } + } + } + finally + { + // Do we want to try to undo the step, and not update the bookmark? + conversation.MessageStore.UpdateBookmark(); + } + } + + private async ValueTask UpdateThreadAsync(IReadOnlyCollection messages, AgentThread? thread = null, CancellationToken cancellation = default) + { + if (thread is null) + { + thread = this.GetNewThread(); + } + + if (thread is not WorkflowThread workflowThread) + { + throw new ArgumentException($"Incompatible thread type: {thread.GetType()} (expecting {typeof(WorkflowThread)})", nameof(thread)); + } + + await workflowThread.MessageStore.AddMessagesAsync(messages, cancellation).ConfigureAwait(false); + return workflowThread; + } + + public override async + Task RunAsync( + IReadOnlyCollection messages, + AgentThread? thread = null, + AgentRunOptions? options = null, + CancellationToken cancellationToken = default) + { + WorkflowThread workflowThread = await this.UpdateThreadAsync(messages, thread, cancellationToken).ConfigureAwait(false); + MessageMerger merger = new(); + + await foreach (AgentRunResponseUpdate update in this.InvokeStageAsync(workflowThread, cancellationToken) + .ConfigureAwait(false) + .WithCancellation(cancellationToken)) + { + merger.AddUpdate(update); + } + + return merger.ComputeMerged(workflowThread.ResponseId); + } + + public override async + IAsyncEnumerable RunStreamingAsync( + IReadOnlyCollection messages, + AgentThread? thread = null, + AgentRunOptions? options = null, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + WorkflowThread workflowThread = await this.UpdateThreadAsync(messages, thread, cancellationToken).ConfigureAwait(false); + await foreach (AgentRunResponseUpdate update in this.InvokeStageAsync(workflowThread, cancellationToken) + .ConfigureAwait(false) + .WithCancellation(cancellationToken)) + { + yield return update; + } + } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/WorkflowHostingExtensions.cs b/dotnet/src/Microsoft.Agents.Workflows/WorkflowHostingExtensions.cs new file mode 100644 index 0000000000..abc679c844 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.Workflows/WorkflowHostingExtensions.cs @@ -0,0 +1,35 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Collections.Generic; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.AI.Agents; + +namespace Microsoft.Agents.Workflows; + +/// +/// Provides extension methods for treating workflows as +/// +public static class WorkflowHostingExtensions +{ + /// + /// Convert a workflow with the appropriate primary input type to an . + /// + /// + /// + /// + /// + public static AIAgent AsAgent(this Workflow> workflow, string? id = null, string? name = null) + { + return new WorkflowHostAgent(workflow, id, name); + } + + internal static FunctionCallContent ToFunctionCall(this ExternalRequest request) + { + Dictionary parameters = new() + { + { "data", request.Data} + }; + + return new FunctionCallContent(request.RequestId, request.Port.Id, parameters); + } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/WorkflowMessageStore.cs b/dotnet/src/Microsoft.Agents.Workflows/WorkflowMessageStore.cs new file mode 100644 index 0000000000..35de73ce0b --- /dev/null +++ b/dotnet/src/Microsoft.Agents.Workflows/WorkflowMessageStore.cs @@ -0,0 +1,90 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Collections.Generic; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.AI.Agents; + +namespace Microsoft.Agents.Workflows; + +internal class WorkflowMessageStore : IChatMessageStore +{ + private int _bookmark = 0; + private readonly List _chatMessages = new(); + + internal class StoreState + { + public int Bookmark { get; set; } + public IList Messages { get; set; } = new List(); + } + + internal void AddMessages(params ChatMessage[] messages) + { + this._chatMessages.AddRange(messages); + } + + public Task AddMessagesAsync(IReadOnlyCollection messages, CancellationToken cancellationToken) + { + this._chatMessages.AddRange(messages); + + return Task.CompletedTask; + } + + public Task> GetMessagesAsync(CancellationToken cancellationToken) + { + return Task.FromResult>(this._chatMessages.AsReadOnly()); + } + + public IEnumerable GetFromBookmark() + { + for (int i = this._bookmark; i < this._chatMessages.Count; i++) + { + yield return this._chatMessages[i]; + } + } + + public void UpdateBookmark() + { + this._bookmark = this._chatMessages.Count; + } + + public ValueTask DeserializeStateAsync(JsonElement? serializedStoreState, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default) + { + if (serializedStoreState is null) + { + return default; + } + + object? maybeState = + JsonSerializer.Deserialize( + serializedStoreState.Value, + AgentAbstractionsJsonUtilities.DefaultOptions.GetTypeInfo(typeof(StoreState))); + + if (maybeState is not StoreState state) + { + throw new JsonException("Invalid state format for WorkflowMessageStore."); + } + + this._chatMessages.Clear(); + this._chatMessages.AddRange(state.Messages); + + this._bookmark = state.Bookmark; + + return default; + } + + public ValueTask SerializeStateAsync(JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default) + { + StoreState state = new() + { + Bookmark = this._bookmark, + Messages = this._chatMessages, + }; + + return new ValueTask + (JsonSerializer.SerializeToElement(state, + WorkflowsJsonUtilities.DefaultOptions.GetTypeInfo(typeof(StoreState)))); + } +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/WorkflowThread.cs b/dotnet/src/Microsoft.Agents.Workflows/WorkflowThread.cs new file mode 100644 index 0000000000..d8261c77e3 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.Workflows/WorkflowThread.cs @@ -0,0 +1,60 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.AI.Agents; +using Microsoft.Shared.Diagnostics; + +namespace Microsoft.Agents.Workflows; + +internal class WorkflowThread : AgentThread +{ + private readonly string _workflowId; + private readonly string? _workflowName; + private readonly WorkflowMessageStore _messageStore; + + public WorkflowThread(string workflowId, string? workflowName, string runId) + { + base.MessageStore = this._messageStore = new(); + this.RunId = Throw.IfNullOrEmpty(runId, nameof(runId)); + + this._workflowId = Throw.IfNullOrEmpty(workflowId); + this._workflowName = workflowName; + } + + public string RunId { get; } + public int Halts { get; } = 0; + + public string ResponseId => $"{this.RunId}@{this.Halts}"; + + public override Task SerializeAsync(JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default) + { + throw new NotImplementedException("Pending Checkpointing work."); + } + + protected override Task DeserializeAsync(JsonElement serializedThread, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default) + { + throw new NotImplementedException("Pending Checkpointing work."); + } + + public AgentRunResponseUpdate CreateUpdate(params AIContent[] parts) + { + Throw.IfNullOrEmpty(parts); + + AgentRunResponseUpdate update = new(ChatRole.Assistant, parts) + { + CreatedAt = DateTimeOffset.Now, + MessageId = Guid.NewGuid().ToString("N"), + }; + + this.MessageStore.AddMessages(update.ToChatMessage()); + + return update; + } + + /// + public new WorkflowMessageStore MessageStore => this._messageStore; +} diff --git a/dotnet/src/Microsoft.Agents.Workflows/WorkflowsJsonUtilities.cs b/dotnet/src/Microsoft.Agents.Workflows/WorkflowsJsonUtilities.cs new file mode 100644 index 0000000000..5f4efc332d --- /dev/null +++ b/dotnet/src/Microsoft.Agents.Workflows/WorkflowsJsonUtilities.cs @@ -0,0 +1,61 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Diagnostics.CodeAnalysis; +using System.Text.Json; +using System.Text.Json.Serialization; +using Microsoft.Extensions.AI; +using static Microsoft.Agents.Workflows.WorkflowMessageStore; + +namespace Microsoft.Agents.Workflows; + +/// Provides a collection of utility methods for working with JSON data in the context of workflows. +internal static partial class WorkflowsJsonUtilities +{ + /// + /// Gets the singleton used as the default in JSON serialization operations. + /// + /// + /// + /// For Native AOT or applications disabling , this instance + /// includes source generated contracts for all common exchange types contained in this library. + /// + /// + /// It additionally turns on the following settings: + /// + /// Enables defaults. + /// Enables as the default ignore condition for properties. + /// Enables as the default number handling for number types. + /// + /// + /// + public static JsonSerializerOptions DefaultOptions { get; } = CreateDefaultOptions(); + + /// + /// Creates default options to use for agents-related serialization. + /// + /// The configured options. + [UnconditionalSuppressMessage("ReflectionAnalysis", "IL3050:RequiresDynamicCode", Justification = "Converter is guarded by IsReflectionEnabledByDefault check.")] + [UnconditionalSuppressMessage("Trimming", "IL2026:Members annotated with 'RequiresUnreferencedCodeAttribute' require dynamic access", Justification = "Converter is guarded by IsReflectionEnabledByDefault check.")] + private static JsonSerializerOptions CreateDefaultOptions() + { + // Copy the configuration from the source generated context. + JsonSerializerOptions options = new(JsonContext.Default.Options); + + // Chain with all supported types from Microsoft.Extensions.AI.Abstractions. and Microsoft.Extensions.AI.Agents.Abstractions. + options.TypeInfoResolverChain.Add(AIJsonUtilities.DefaultOptions.TypeInfoResolver!); + + options.MakeReadOnly(); + return options; + } + + // Keep in sync with CreateDefaultOptions above. + [JsonSourceGenerationOptions(JsonSerializerDefaults.Web, + DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull, + NumberHandling = JsonNumberHandling.AllowReadingFromString)] + + // Agent abstraction types + [JsonSerializable(typeof(StoreState))] + + [ExcludeFromCodeCoverage] + internal sealed partial class JsonContext : JsonSerializerContext; +} diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/06_GroupChat_Workflow.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/06_GroupChat_Workflow.cs index 02ebc76ab0..d631570902 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/06_GroupChat_Workflow.cs +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/06_GroupChat_Workflow.cs @@ -5,6 +5,7 @@ using System.Collections.Generic; using System.Diagnostics; using System.IO; using System.Linq; +using System.Runtime.CompilerServices; using System.Text; using System.Threading; using System.Threading.Tasks; @@ -15,17 +16,20 @@ namespace Microsoft.Agents.Workflows.UnitTests.Sample; internal static class Step6EntryPoint { - internal static int MaxSteps { get; set; } + public static Workflow> CreateWorkflow(int maxTurns) + { + GroupChatBuilder builder = + GroupChatBuilder.Create + (options => options.MaxTurns = maxTurns) + .AddParticipant(new HelloAgent(), shouldEmitEvents: true) + .AddParticipant(new EchoAgent(), shouldEmitEvents: true); + + return builder.ReduceToWorkflow(); + } public static async ValueTask RunAsync(TextWriter writer, int maxSteps = 2) { - Step6EntryPoint.MaxSteps = maxSteps; - - GroupChatBuilder builder = GroupChatBuilder.Create() - .AddParticipant(new HelloAgent(), shouldEmitEvents: true) - .AddParticipant(new EchoAgent(), shouldEmitEvents: true); - - Workflow> workflow = builder.ReduceToWorkflow(); + Workflow> workflow = CreateWorkflow(maxSteps); StreamingRun run = await InProcessExecution.StreamAsync(workflow, []) .ConfigureAwait(false); @@ -37,20 +41,34 @@ internal static class Step6EntryPoint { Debug.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}"); } - else if (evt is AgentRunEvent agentRun && agentRun.Data is AgentRunResponse response) + else if (evt is AgentRunUpdateEvent update) { + AgentRunResponse response = update.AsResponse(); + foreach (ChatMessage message in response.Messages) { - writer.WriteLine($"{agentRun.ExecutorId}: {message.Text}"); + writer.WriteLine($"{update.ExecutorId}: {message.Text}"); } } } } - private sealed class RoundRobinGroupChatManager : GroupChatManager + private sealed class RoundRobinGroupChatManagerOptions : GroupChatManagerOptions + { + public int? MaxTurns { get; set; } = null; + } + + private sealed class RoundRobinGroupChatManager() : GroupChatManager { public int TurnCount { get; private set; } = 0; - public int MaxTurns { get; init; } = Step6EntryPoint.MaxSteps; + public int? MaxTurns { get; private set; } = null; + + protected internal override void Configure(RoundRobinGroupChatManagerOptions options) + { + base.Configure(options); + + this.MaxTurns = options.MaxTurns; + } public override int? GetNextTurnExecutor(GroupChatHistory history) { @@ -75,17 +93,27 @@ internal sealed class HelloAgent(string id = nameof(HelloAgent)) : AIAgent public const string DefaultId = nameof(HelloAgent); public override string Id => id; + public override string? Name => id; - public override Task RunAsync(IReadOnlyCollection messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default) + public override async Task RunAsync(IReadOnlyCollection messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default) { - AgentRunResponse response = new(new ChatMessage(ChatRole.Assistant, "Hello World!")); + IEnumerable update = [ + await this.RunStreamingAsync(messages, thread, options, cancellationToken) + .SingleAsync(cancellationToken) + .ConfigureAwait(false)]; - return Task.FromResult(response); + return update.ToAgentRunResponse(); } - public override IAsyncEnumerable RunStreamingAsync(IReadOnlyCollection messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default) + public override async IAsyncEnumerable RunStreamingAsync(IReadOnlyCollection messages, AgentThread? thread = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) { - throw new NotImplementedException(); + AgentRunResponseUpdate response = new(ChatRole.Assistant, "Hello World!") + { + AgentId = this.Id, + AuthorName = this.Name, + }; + + yield return response; } } @@ -95,8 +123,19 @@ internal sealed class EchoAgent(string id = nameof(EchoAgent)) : AIAgent public const string DefaultId = nameof(EchoAgent); public override string Id => id; + public override string? Name => id; - public override Task RunAsync(IReadOnlyCollection messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default) + public override async Task RunAsync(IReadOnlyCollection messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default) + { + IEnumerable update = [ + await this.RunStreamingAsync(messages, thread, options, cancellationToken) + .SingleAsync(cancellationToken) + .ConfigureAwait(false)]; + + return update.ToAgentRunResponse(); + } + + public override async IAsyncEnumerable RunStreamingAsync(IReadOnlyCollection messages, AgentThread? thread = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) { if (messages.Count == 0) { @@ -110,13 +149,13 @@ internal sealed class EchoAgent(string id = nameof(EchoAgent)) : AIAgent collectedText.AppendLine(messageText); } - AgentRunResponse result = new(new ChatMessage(ChatRole.Assistant, collectedText.ToString())); - return Task.FromResult(result); - } + AgentRunResponseUpdate result = new(ChatRole.Assistant, collectedText.ToString()) + { + AgentId = this.Id, + AuthorName = this.Name, + }; - public override IAsyncEnumerable RunStreamingAsync(IReadOnlyCollection messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default) - { - throw new NotImplementedException(); + yield return result; } } @@ -144,6 +183,10 @@ internal sealed class GroupChatHistory public IEnumerable NewMessagesThisTurn => this._messages.Skip(this._bookmark); } +internal class GroupChatManagerOptions +{ +} + internal abstract class GroupChatManager { public string[] ParticipantIds { get; internal init; } = []; @@ -151,6 +194,11 @@ internal abstract class GroupChatManager public abstract int? GetNextTurnExecutor(GroupChatHistory history); } +internal abstract class GroupChatManager : GroupChatManager where TOptions : GroupChatManagerOptions, new() +{ + protected internal virtual void Configure(TOptions options) { } +} + internal sealed class GroupChatBuilder { private readonly List _participants = new(); @@ -167,6 +215,21 @@ internal sealed class GroupChatBuilder return new GroupChatBuilder(participantIds => new TManager() { ParticipantIds = participantIds }); } + public static GroupChatBuilder Create(Action configure) + where TManager : GroupChatManager, new() + where TOptions : GroupChatManagerOptions, new() + { + TOptions options = new(); + configure(options); + + return new GroupChatBuilder(participantIds => + { + TManager manager = new() { ParticipantIds = participantIds }; + manager.Configure(options); + return manager; + }); + } + public GroupChatBuilder AddParticipant(ExecutorIsh executor, bool shouldEmitEvents = false) { this._participants.Add(executor); @@ -265,7 +328,7 @@ internal sealed class GroupChatBuilder if (this.TryEnterConversation()) { // Capture the initial turn token's EmitEvents setting - this._shouldHostEmitEvents = token.EmitEvents; + this._shouldHostEmitEvents = token.EmitEvents.HasValue ? token.EmitEvents.Value : false; } int? nextSpeakerIndex = this._manager.GetNextTurnExecutor(this._history); diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs new file mode 100644 index 0000000000..9bff976f35 --- /dev/null +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs @@ -0,0 +1,31 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Collections.Generic; +using System.IO; +using System.Threading.Tasks; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.AI.Agents; + +namespace Microsoft.Agents.Workflows.UnitTests.Sample; + +internal static class Step7EntryPoint +{ + public static async ValueTask RunAsync(TextWriter writer, int maxSteps = 2) + { + Workflow> workflow = Step6EntryPoint.CreateWorkflow(maxSteps); + AIAgent agent = workflow.AsAgent("group-chat-agent", "Group Chat Agent"); + + AgentThread thread = agent.GetNewThread(); + + await foreach (AgentRunResponseUpdate update in agent.RunStreamingAsync(thread).ConfigureAwait(false)) + { + string updateText = $"{update.AuthorName + ?? update.AgentId + ?? update.Role.ToString() + ?? ChatRole.Assistant.ToString()}: {update.Text}"; + Console.WriteLine(updateText); + writer.WriteLine(updateText); + } + } +} diff --git a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/SampleSmokeTest.cs b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/SampleSmokeTest.cs index 7eca842077..0ac7635533 100644 --- a/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/SampleSmokeTest.cs +++ b/dotnet/tests/Microsoft.Agents.Workflows.UnitTests/SampleSmokeTest.cs @@ -100,6 +100,22 @@ public class SampleSmokeTest line => Assert.Contains($"{EchoAgent.DefaultId}: {EchoAgent.Prefix}{HelloAgent.Greeting}", line) ); } + + [Fact] + public async Task Test_RunSample_Step7Async() + { + using StringWriter writer = new(); + + await Step7EntryPoint.RunAsync(writer); + + string result = writer.ToString(); + string[] lines = result.Split([Environment.NewLine], StringSplitOptions.RemoveEmptyEntries); + + Assert.Collection(lines, + line => Assert.Contains($"{HelloAgent.DefaultId}: {HelloAgent.Greeting}", line), + line => Assert.Contains($"{EchoAgent.DefaultId}: {EchoAgent.Prefix}{HelloAgent.Greeting}", line) + ); + } } internal sealed class VerifyingPlaybackResponder