// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.AI;
namespace Microsoft.Agents.AI.Workflows;
///
/// Provides configuration options for .
///
public class ChatProtocolExecutorOptions
{
///
/// Gets or sets the chat role to use when converting string messages to instances.
/// If set, the executor will accept string messages and convert them to chat messages with this role.
///
public ChatRole? StringMessageChatRole { get; set; }
///
/// Gets or sets a value indicating whether the executor should automatically send the
/// after returning from
///
public bool AutoSendTurnToken { get; set; } = true;
}
///
/// Provides a base class for executors that implement the Agent Workflow Chat Protocol.
/// This executor maintains a list of chat messages and processes them when a turn is taken.
///
public abstract class ChatProtocolExecutor : StatefulExecutor>
{
internal static readonly Func> s_initFunction = () => [];
private readonly ChatProtocolExecutorOptions _options;
private static readonly StatefulExecutorOptions s_baseExecutorOptions = new()
{
AutoSendMessageHandlerResultObject = false,
AutoYieldOutputHandlerResultObject = false
};
///
/// Initializes a new instance of the class.
///
/// The unique identifier for this executor instance. Cannot be null or empty.
/// Optional configuration settings for the executor. If null, default options are used.
/// Declare that this executor may be used simultaneously by multiple runs safely.
protected ChatProtocolExecutor(string id, ChatProtocolExecutorOptions? options = null, bool declareCrossRunShareable = false)
: base(id, () => [], s_baseExecutorOptions, declareCrossRunShareable)
{
this._options = options ?? new();
}
///
/// Gets a value indicating whether string-based messages are supported by this .
///
[MemberNotNullWhen(true, nameof(StringMessageChatRole))]
protected bool SupportsStringMessage => this.StringMessageChatRole.HasValue;
///
protected ChatRole? StringMessageChatRole => this._options.StringMessageChatRole;
///
protected bool AutoSendTurnToken => this._options.AutoSendTurnToken;
///
protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)
{
return protocolBuilder.ConfigureRoutes(ConfigureRoutes)
.SendsMessage>()
.SendsMessage();
void ConfigureRoutes(RouteBuilder routeBuilder)
{
if (this.SupportsStringMessage)
{
routeBuilder = routeBuilder.AddHandler(
(message, context) => this.AddMessageAsync(new(this.StringMessageChatRole.Value, message), context));
}
routeBuilder.AddHandler(this.AddMessageAsync)
.AddHandler>(this.AddMessagesAsync)
.AddHandler(this.AddMessagesAsync)
//.AddHandler>(this.AddMessagesAsync)
.AddHandler(this.TakeTurnAsync);
}
}
///
/// Adds a single chat message to the accumulated messages for the current turn.
///
/// The chat message to add.
/// The workflow context in which the executor executes.
/// The to monitor for cancellation requests.
/// A representing the asynchronous operation.
protected ValueTask AddMessageAsync(ChatMessage message, IWorkflowContext context, CancellationToken cancellationToken = default)
{
return this.InvokeWithStateAsync(ForwardMessageAsync, context, cancellationToken: cancellationToken);
ValueTask?> ForwardMessageAsync(List? maybePendingMessages, IWorkflowContext context, CancellationToken cancelationToken)
{
maybePendingMessages ??= s_initFunction();
maybePendingMessages.Add(message);
return new(maybePendingMessages);
}
}
///
/// Adds multiple chat messages to the accumulated messages for the current turn.
///
/// The collection of chat messages to add.
/// The workflow context in which the executor executes.
/// The to monitor for cancellation requests.
/// A representing the asynchronous operation.
protected ValueTask AddMessagesAsync(IEnumerable messages, IWorkflowContext context, CancellationToken cancellationToken = default)
{
return this.InvokeWithStateAsync(ForwardMessageAsync, context, cancellationToken: cancellationToken);
ValueTask?> ForwardMessageAsync(List? maybePendingMessages, IWorkflowContext context, CancellationToken cancelationToken)
{
maybePendingMessages ??= s_initFunction();
maybePendingMessages.AddRange(messages);
return new(maybePendingMessages);
}
}
///
/// Handles a turn token by processing all accumulated chat messages and then resetting the message state.
///
/// The turn token that triggers message processing.
/// The workflow context in which the executor executes.
/// The to monitor for cancellation requests.
/// A representing the asynchronous operation.
public ValueTask TakeTurnAsync(TurnToken token, IWorkflowContext context, CancellationToken cancellationToken = default)
{
return this.InvokeWithStateAsync(InvokeTakeTurnAsync, context, cancellationToken: cancellationToken);
async ValueTask?> InvokeTakeTurnAsync(List? maybePendingMessages, IWorkflowContext context, CancellationToken cancellationToken)
{
await this.TakeTurnAsync(maybePendingMessages ?? s_initFunction(), context, token.EmitEvents, cancellationToken)
.ConfigureAwait(false);
if (this.AutoSendTurnToken)
{
await context.SendMessageAsync(token, cancellationToken: cancellationToken).ConfigureAwait(false);
}
// Rerun the initialStateFactory to reset the state to empty list. (We could return the empty list directly,
// but this is more consistent if the initial state factory becomes more complex.)
return s_initFunction();
}
}
///
/// Processes the current set of turn messages using the specified asynchronous processing function.
///
/// If the provided list of chat messages is null, an initial empty list is supplied to the
/// processing function. If the processing function returns null, an empty list is used as the result.
/// A delegate that asynchronously processes a list of chat messages within the given workflow context and
/// cancellation token, returning the processed list of chat messages or null.
/// The workflow context in which the messages are processed.
/// A token that can be used to cancel the asynchronous operation.
/// A ValueTask that represents the asynchronous operation. The result contains the processed list of chat messages,
/// or an empty list if the processing function returns null.
protected ValueTask ProcessTurnMessagesAsync(Func, IWorkflowContext, CancellationToken, ValueTask?>> processFunc, IWorkflowContext context, CancellationToken cancellationToken)
{
return this.InvokeWithStateAsync(InvokeProcessFuncAsync, context, cancellationToken: cancellationToken);
async ValueTask?> InvokeProcessFuncAsync(List? maybePendingMessages, IWorkflowContext context, CancellationToken cancellationToken)
{
return (await processFunc(maybePendingMessages ?? s_initFunction(), context, cancellationToken).ConfigureAwait(false))
?? s_initFunction();
}
}
///
/// When overridden in a derived class, processes the accumulated chat messages for a single turn.
///
/// The list of chat messages accumulated since the last turn.
/// The workflow context in which the executor executes.
/// Indicates whether events should be emitted during processing. If null, the default behavior is used.
/// The to monitor for cancellation requests.
/// A representing the asynchronous operation.
protected abstract ValueTask TakeTurnAsync(List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellationToken = default);
}