agent/agent-framework
2
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

.NET: Update Workflow Input/Output Redesign (#881)

Browse Source 
* feat: Make Executor id field mandatory

When checkpointing is involved, it is critical to keep executor ids consistent between runs, even when recreating a new object tree for the workflow.

The default id-setting mechanism generated a guid for part of the id, making it not work when restoring from a checkpoint.

This change prevents this situation from arising.

* feat: Enable running untyped Workflows

With the change to enable delay-instantiation of executors and support for async Executor factory methods, we must instantiate the starting executor to know what are the valid input types for the workflow.

To avoid forcing instantiation every time, and to better support workflows with multiple input types, we enable support for build and interacting with the base Workflow type without type annotations, and remove the requirement to know a valid input type when initiating a run.

* feat: Support Output from any executor and multiple outputs.
This commit is contained in:
Jacob Alber
2025-09-24 22:03:22 -04:00
committed by GitHub
Unverified
parent 03ef7f054f
commit 39e071c430
89 changed files with 1413 additions and 998 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dotnet/src/Microsoft.Agents.Workflows.Declarative/DeclarativeWorkflowBuilder.cs
+3 -3
View File
@@ -24,7 +24,7 @@ public static class DeclarativeWorkflowBuilder
/// <param name="options">Configuration options for workflow execution.</param>
/// <param name="inputTransform">An optional function to transform the input message into a <see cref="ChatMessage"/>.</param>
/// <returns></returns>
public static Workflow<TInput> Build<TInput>(
public static Workflow Build<TInput>(
string workflowFile,
DeclarativeWorkflowOptions options,
Func<TInput, ChatMessage>? inputTransform = null)
@@ -42,7 +42,7 @@ public static class DeclarativeWorkflowBuilder
/// <param name="options">Configuration options for workflow execution.</param>
/// <param name="inputTransform">An optional function to transform the input message into a <see cref="ChatMessage"/>.</param>
/// <returns>The <see cref="Workflow"/> that corresponds with the YAML object model.</returns>
public static Workflow<TInput> Build<TInput>(
public static Workflow Build<TInput>(
TextReader yamlReader,
DeclarativeWorkflowOptions options,
Func<TInput, ChatMessage>? inputTransform = null)
@@ -68,7 +68,7 @@ public static class DeclarativeWorkflowBuilder
WorkflowElementWalker walker = new(visitor);
walker.Visit(rootElement);
return visitor.Complete<TInput>();
return visitor.Complete();
}
private static ChatMessage DefaultTransform(object message) =>
dotnet/src/Microsoft.Agents.Workflows.Declarative/Interpreter/DeclarativeWorkflowContext.cs
+6
View File
@@ -32,6 +32,12 @@ internal sealed class DeclarativeWorkflowContext : IWorkflowContext
/// <inheritdoc/>
public ValueTask AddEventAsync(WorkflowEvent workflowEvent) => this.Source.AddEventAsync(workflowEvent);
/// <inheritdoc/>
public ValueTask YieldOutputAsync(object output) => this.Source.YieldOutputAsync(output);
/// <inheritdoc/>
public ValueTask RequestHaltAsync() => this.Source.RequestHaltAsync();
/// <inheritdoc/>
public async ValueTask QueueClearScopeAsync(string? scopeName = null)
{
dotnet/src/Microsoft.Agents.Workflows.Declarative/Interpreter/WorkflowActionVisitor.cs
+2 -2
View File
@@ -40,13 +40,13 @@ internal sealed class WorkflowActionVisitor : DialogActionVisitor
public bool HasUnsupportedActions { get; private set; }
public Workflow<TInput> Complete<TInput>()
public Workflow Complete()
{
// Process the cached links
this._workflowModel.ConnectNodes(this._workflowBuilder);
// Build final workflow
return this._workflowBuilder.Build<TInput>();
return this._workflowBuilder.Build();
}
protected override void Visit(ActionScope item)
dotnet/src/Microsoft.Agents.Workflows/AgentWorkflowBuilder.cs
+33 -50
View File
@@ -26,7 +26,7 @@ public static partial class AgentWorkflowBuilder
/// </summary>
/// <param name="agents">The sequence of agents to compose into a sequential workflow.</param>
/// <returns>The built workflow composed of the supplied <paramref name="agents"/>, in the order in which they were yielded from the source.</returns>
public static Workflow<List<ChatMessage>> BuildSequential(params IEnumerable<AIAgent> agents)
public static Workflow BuildSequential(params IEnumerable<AIAgent> agents)
{
Throw.IfNull(agents);
@@ -59,9 +59,11 @@ public static partial class AgentWorkflowBuilder
// Add an ending executor that batches up all messages from the last agent
// so that it's published as a single list result.
Debug.Assert(builder is not null);
builder.AddEdge(previous, new ConvertMessageListToCompletedEventExecutor());
return builder.Build<List<ChatMessage>>();
OutputMessagesExecutor end = new();
return builder.AddEdge(previous, end)
.WithOutputFrom(end)
.Build();
}
/// <summary>
@@ -75,14 +77,14 @@ public static partial class AgentWorkflowBuilder
/// from each agent that produced at least one message.
/// </param>
/// <returns>The built workflow composed of the supplied concurrent <paramref name="agents"/>.</returns>
public static Workflow<List<ChatMessage>> BuildConcurrent(
public static Workflow BuildConcurrent(
IEnumerable<AIAgent> agents,
Func<IList<List<ChatMessage>>, List<ChatMessage>>? aggregator = null)
{
Throw.IfNull(agents);
// A workflow needs a starting executor, so we create one that forwards everything to each agent.
ForwardingExecutor start = new();
ChatForwardingExecutor start = new("Start");
WorkflowBuilder builder = new(start);
// For each agent, we create an executor to host it and an accumulator to batch up its output messages,
@@ -90,7 +92,7 @@ public static partial class AgentWorkflowBuilder
// accumulator would not be able to determine what came from what agent, as there's currently no
// provenance tracking exposed in the workflow context passed to a handler.
ExecutorIsh[] agentExecutors = (from agent in agents select (ExecutorIsh)new AgentRunStreamingExecutor(agent, includeInputInOutput: false)).ToArray();
ExecutorIsh[] accumulators = [.. from agent in agentExecutors select (ExecutorIsh)new BatchChatMessagesToListExecutor()];
ExecutorIsh[] accumulators = [.. from agent in agentExecutors select (ExecutorIsh)new BatchChatMessagesToListExecutor($"Batcher/{agent.Id}")];
builder.AddFanOutEdge(start, targets: agentExecutors);
for (int i = 0; i < agentExecutors.Length; i++)
{
@@ -104,7 +106,7 @@ public static partial class AgentWorkflowBuilder
ConcurrentEndExecutor end = new(agentExecutors.Length, aggregator);
builder.AddFanInEdge(end, sources: accumulators);
return builder.Build<List<ChatMessage>>();
return builder.WithOutputFrom(end).Build();
}
/// <summary>Creates a new <see cref="HandoffsWorkflowBuilder"/> using <paramref name="initialAgent"/> as the starting agent in the workflow.</summary>
@@ -189,49 +191,31 @@ public static partial class AgentWorkflowBuilder
/// Provides an executor that batches received chat messages that it then publishes as the final result
/// when receiving a <see cref="TurnToken"/>.
/// </summary>
private sealed class ConvertMessageListToCompletedEventExecutor : Executor
private sealed class OutputMessagesExecutor() : ChatProtocolExecutor("OutputMessages")
{
private readonly List<ChatMessage> _pendingMessages = [];
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) =>
routeBuilder
.AddHandler<ChatMessage>((message, context) => this._pendingMessages.Add(message))
.AddHandler<List<ChatMessage>>((messages, _) => this._pendingMessages.AddRange(messages))
.AddHandler<TurnToken>(async (token, context) =>
{
var messages = new List<ChatMessage>(this._pendingMessages);
this._pendingMessages.Clear();
await context.AddEventAsync(new WorkflowCompletedEvent(messages)).ConfigureAwait(false);
});
protected override ValueTask TakeTurnAsync(List<ChatMessage> messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellation = default)
=> context.YieldOutputAsync(messages);
}
/// <summary>Executor that forwards all messages.</summary>
private sealed class ForwardingExecutor : Executor
private sealed class ChatForwardingExecutor(string id) : Executor(id)
{
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) =>
routeBuilder.AddHandler<object>((message, context) => context.SendMessageAsync(message));
routeBuilder
.AddHandler<string>((message, context) => context.SendMessageAsync(new ChatMessage(ChatRole.User, message)))
.AddHandler<ChatMessage>((message, context) => context.SendMessageAsync(message))
.AddHandler<List<ChatMessage>>((messages, context) => context.SendMessageAsync(messages))
.AddHandler<TurnToken>((turnToken, context) => context.SendMessageAsync(turnToken));
}
/// <summary>
/// Provides an executor that batches received chat messages that it then releases when
/// receiving a <see cref="TurnToken"/>.
/// </summary>
private sealed class BatchChatMessagesToListExecutor : Executor
private sealed class BatchChatMessagesToListExecutor(string id) : ChatProtocolExecutor(id)
{
private readonly List<ChatMessage> _pendingMessages = [];
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) =>
routeBuilder
.AddHandler<ChatMessage>((message, context) => this._pendingMessages.Add(message))
.AddHandler<List<ChatMessage>>((messages, _) => this._pendingMessages.AddRange(messages))
.AddHandler<TurnToken>(async (token, context) =>
{
var messages = new List<ChatMessage>(this._pendingMessages);
this._pendingMessages.Clear();
await context.SendMessageAsync(messages).ConfigureAwait(false);
await context.SendMessageAsync(token).ConfigureAwait(false);
});
protected override ValueTask TakeTurnAsync(List<ChatMessage> messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellation = default)
=> context.SendMessageAsync(messages);
}
/// <summary>
@@ -245,7 +229,7 @@ public static partial class AgentWorkflowBuilder
private List<List<ChatMessage>> _allResults;
private int _remaining;
public ConcurrentEndExecutor(int expectedInputs, Func<IList<List<ChatMessage>>, List<ChatMessage>> aggregator)
public ConcurrentEndExecutor(int expectedInputs, Func<IList<List<ChatMessage>>, List<ChatMessage>> aggregator) : base("ConcurrentEnd")
{
this._expectedInputs = expectedInputs;
this._aggregator = Throw.IfNull(aggregator);
@@ -272,8 +256,7 @@ public static partial class AgentWorkflowBuilder
var results = this._allResults;
this._allResults = new List<List<ChatMessage>>(this._expectedInputs);
await context.AddEventAsync(new WorkflowCompletedEvent(this._aggregator(results))).ConfigureAwait(false);
await context.YieldOutputAsync(this._aggregator(results)).ConfigureAwait(false);
}
});
}
@@ -414,7 +397,7 @@ public static partial class AgentWorkflowBuilder
/// agent to process messages selected by the current agent.
/// </summary>
/// <returns>The workflow built based on the handoffs in the builder.</returns>
public Workflow<List<ChatMessage>> Build()
public Workflow Build()
{
StartHandoffsExecutor start = new();
EndHandoffsExecutor end = new();
@@ -434,7 +417,7 @@ public static partial class AgentWorkflowBuilder
}
// Build the workflow.
return builder.Build<List<ChatMessage>>();
return builder.WithOutputFrom(end).Build();
}
/// <summary>Describes a handoff to a specific target <see cref="AIAgent"/>.</summary>
@@ -445,7 +428,7 @@ public static partial class AgentWorkflowBuilder
}
/// <summary>Executor used at the start of a handoffs workflow to accumulate messages and emit them as HandoffState upon receiving a turn token.</summary>
private sealed class StartHandoffsExecutor : Executor
private sealed class StartHandoffsExecutor() : Executor("HandoffStart")
{
private readonly List<ChatMessage> _pendingMessages = [];
@@ -465,11 +448,11 @@ public static partial class AgentWorkflowBuilder
}
/// <summary>Executor used at the end of a handoff workflow to raise a final completed event.</summary>
private sealed class EndHandoffsExecutor : Executor
private sealed class EndHandoffsExecutor() : Executor("HandoffEnd")
{
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) =>
routeBuilder.AddHandler<HandoffState>((handoff, context) =>
context.AddEventAsync(new WorkflowCompletedEvent(handoff.Messages)));
context.YieldOutputAsync(handoff.Messages));
}
/// <summary>Executor used to represent an agent in a handoffs workflow, responding to <see cref="HandoffState"/> events.</summary>
@@ -740,11 +723,11 @@ public static partial class AgentWorkflowBuilder
}
/// <summary>
/// Builds a <see cref="Workflow{T}"/> composed of agents that operate via group chat, with the next
/// Builds a <see cref="Workflow"/> composed of agents that operate via group chat, with the next
/// agent to process messages selected by the group chat manager.
/// </summary>
/// <returns>The workflow built based on the group chat in the builder.</returns>
public Workflow<List<ChatMessage>> Build()
public Workflow Build()
{
AIAgent[] agents = this._participants.ToArray();
Dictionary<AIAgent, ExecutorIsh> agentMap = agents.ToDictionary(a => a, a => (ExecutorIsh)new AgentRunStreamingExecutor(a, includeInputInOutput: true));
@@ -760,10 +743,10 @@ public static partial class AgentWorkflowBuilder
.AddEdge(participant, host);
}
return builder.Build<List<ChatMessage>>();
return builder.WithOutputFrom(host).Build();
}
private sealed class GroupChatHost(AIAgent[] agents, Dictionary<AIAgent, ExecutorIsh> agentMap, Func<IReadOnlyList<AIAgent>, GroupChatManager> managerFactory) : Executor
private sealed class GroupChatHost(AIAgent[] agents, Dictionary<AIAgent, ExecutorIsh> agentMap, Func<IReadOnlyList<AIAgent>, GroupChatManager> managerFactory) : Executor("GroupChatHost")
{
private readonly AIAgent[] _agents = agents;
private readonly Dictionary<AIAgent, ExecutorIsh> _agentMap = agentMap;
@@ -801,7 +784,7 @@ public static partial class AgentWorkflowBuilder
}
this._manager = null;
await context.AddEventAsync(new WorkflowCompletedEvent(messages)).ConfigureAwait(false);
await context.YieldOutputAsync(messages).ConfigureAwait(false);
});
}
}
dotnet/src/Microsoft.Agents.Workflows/AggregatingExecutor.cs
+52
View File
@@ -0,0 +1,52 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Threading;
using System.Threading.Tasks;
namespace Microsoft.Agents.Workflows;
/// <summary>
/// Executes a workflow step that incrementally aggregates input messages using a user-provided aggregation function.
/// </summary>
/// <remarks>The aggregate state is persisted and restored automatically during workflow checkpointing. This
/// executor is suitable for scenarios where stateful, incremental aggregation of messages is required, such as running
/// totals or event accumulation.</remarks>
/// <typeparam name="TInput">The type of input messages to be processed and aggregated.</typeparam>
/// <typeparam name="TAggregate">The type representing the aggregate state produced by the aggregator function.</typeparam>
/// <param name="id">The unique identifier for this executor instance.</param>
/// <param name="aggregator">A function that computes the new aggregate state from the previous aggregate and the current input message. The
/// function receives the current aggregate (or null if this is the first message) and the input message, and returns
/// the updated aggregate.</param>
/// <param name="options">Optional configuration settings for the executor. If null, default options are used.</param>
/// <seealso cref="StreamingAggregators"/>
public class AggregatingExecutor<TInput, TAggregate>(string id,
Func<TAggregate?, TInput, TAggregate?> aggregator,
ExecutorOptions? options = null) : Executor<TInput, TAggregate?>(id, options)
{
private const string AggregateStateKey = "Aggregate";
private TAggregate? _runningAggregate;
/// <inheritdoc/>
public override ValueTask<TAggregate?> HandleAsync(TInput message, IWorkflowContext context)
{
this._runningAggregate = aggregator(this._runningAggregate, message);
return new(this._runningAggregate);
}
/// <inheritdoc/>
protected internal override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default)
{
await context.QueueStateUpdateAsync(AggregateStateKey, this._runningAggregate).ConfigureAwait(false);
await base.OnCheckpointingAsync(context, cancellation).ConfigureAwait(false);
}
/// <inheritdoc/>
protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default)
{
await base.OnCheckpointRestoredAsync(context, cancellation).ConfigureAwait(false);
this._runningAggregate = await context.ReadStateAsync<TAggregate>(AggregateStateKey).ConfigureAwait(false);
}
}
dotnet/src/Microsoft.Agents.Workflows/ChatProtocolExecutor.cs
+64
View File
@@ -0,0 +1,64 @@
// 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;
namespace Microsoft.Agents.Workflows;
internal class ChatProtocolExecutorOptions
{
public ChatRole? StringMessageChatRole { get; set; }
}
internal abstract class ChatProtocolExecutor(string id, ChatProtocolExecutorOptions? options = null) : Executor(id)
{
private List<ChatMessage> _pendingMessages = [];
private readonly ChatRole? _stringMessageChatRole = options?.StringMessageChatRole;
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder)
{
if (this._stringMessageChatRole.HasValue)
{
routeBuilder = routeBuilder.AddHandler<string>((message, _) => this._pendingMessages.Add(new(this._stringMessageChatRole.Value, message)));
}
return routeBuilder.AddHandler<ChatMessage>((message, _) => this._pendingMessages.Add(message))
.AddHandler<List<ChatMessage>>((messages, _) => this._pendingMessages.AddRange(messages))
.AddHandler<TurnToken>(this.TakeTurnAsync);
}
public async ValueTask TakeTurnAsync(TurnToken token, IWorkflowContext context)
{
await this.TakeTurnAsync(this._pendingMessages, context, token.EmitEvents).ConfigureAwait(false);
this._pendingMessages = new();
await context.SendMessageAsync(token).ConfigureAwait(false);
}
protected abstract ValueTask TakeTurnAsync(List<ChatMessage> messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellation = default);
private const string PendingMessagesStateKey = nameof(_pendingMessages);
protected internal override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default)
{
Task messagesTask = Task.CompletedTask;
if (this._pendingMessages.Count > 0)
{
JsonElement messagesValue = this._pendingMessages.Serialize();
messagesTask = context.QueueStateUpdateAsync(PendingMessagesStateKey, messagesValue).AsTask();
}
await messagesTask.ConfigureAwait(false);
}
protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default)
{
JsonElement? messagesValue = await context.ReadStateAsync<JsonElement?>(PendingMessagesStateKey).ConfigureAwait(false);
if (messagesValue.HasValue)
{
List<ChatMessage> messages = messagesValue.Value.DeserializeMessages();
this._pendingMessages.AddRange(messages);
}
}
}
dotnet/src/Microsoft.Agents.Workflows/Checkpointed.cs
-4
View File
@@ -13,9 +13,7 @@ namespace Microsoft.Agents.Workflows;
/// </summary>
/// <typeparam name="TRun">The type of the underlying workflow run handle.</typeparam>
/// <seealso cref="Run"/>
/// <seealso cref="Run{TResult}"/>
/// <seealso cref="StreamingRun"/>
/// <seealso cref="StreamingRun{TResult}"/>
public class Checkpointed<TRun>
{
private readonly ICheckpointingRunner _runner;
@@ -30,9 +28,7 @@ public class Checkpointed<TRun>
/// Gets the workflow run associated with this <see cref="Checkpointed{TRun}"/> instance.
/// </summary>
/// <seealso cref="Run"/>
/// <seealso cref="Run{TResult}"/>
/// <seealso cref="StreamingRun"/>
/// <seealso cref="StreamingRun{TResult}"/>
public TRun Run { get; }
/// <inheritdoc cref="ICheckpointingRunner.Checkpoints"/>
dotnet/src/Microsoft.Agents.Workflows/Checkpointing/RepresentationExtensions.cs
+6 -6
View File
@@ -33,7 +33,7 @@ internal static class RepresentationExtensions
return new(new TypeId(port.Request), new TypeId(port.Response), port.Id);
}
private static WorkflowInfo ToWorkflowInfo<TInput>(this Workflow<TInput> workflow, TypeId? outputType, string? outputExecutorId)
private static WorkflowInfo ToWorkflowInfo(this Workflow workflow, TypeId? inputType, TypeId? outputType, string? outputExecutorId)
{
Throw.IfNull(workflow);
@@ -48,12 +48,12 @@ internal static class RepresentationExtensions
HashSet<InputPortInfo> inputPorts = new(workflow.Ports.Values.Select(ToPortInfo));
return new WorkflowInfo(executors, edges, inputPorts, new TypeId(workflow.InputType), workflow.StartExecutorId, outputType, outputExecutorId);
return new WorkflowInfo(executors, edges, inputPorts, inputType, workflow.StartExecutorId, workflow.OutputExecutors);
}
public static WorkflowInfo ToWorkflowInfo<TInput>(this Workflow<TInput> workflow)
=> workflow.ToWorkflowInfo(outputType: null, outputExecutorId: null);
public static WorkflowInfo ToWorkflowInfo(this Workflow workflow)
=> workflow.ToWorkflowInfo(inputType: null, outputType: null, outputExecutorId: null);
public static WorkflowInfo ToWorkflowInfo<TInput, TResult>(this Workflow<TInput, TResult> workflow)
=> workflow.ToWorkflowInfo(outputType: new TypeId(typeof(TResult)), outputExecutorId: workflow.OutputCollectorId);
public static WorkflowInfo ToWorkflowInfo<TInput>(this Workflow<TInput> workflow)
=> workflow.ToWorkflowInfo(inputType: new(workflow.InputType), outputType: null, outputExecutorId: null);
}
dotnet/src/Microsoft.Agents.Workflows/Checkpointing/WorkflowInfo.cs
+20 -31
View File
@@ -1,6 +1,5 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text.Json.Serialization;
@@ -15,53 +14,35 @@ internal sealed class WorkflowInfo
Dictionary<string, ExecutorInfo> executors,
Dictionary<string, List<EdgeInfo>> edges,
HashSet<InputPortInfo> inputPorts,
TypeId inputType,
TypeId? inputType,
string startExecutorId,
TypeId? outputType,
string? outputCollectorId)
HashSet<string>? outputExecutorIds)
{
this.Executors = Throw.IfNull(executors);
this.Edges = Throw.IfNull(edges);
this.InputPorts = Throw.IfNull(inputPorts);
this.InputType = Throw.IfNull(inputType);
this.InputType = inputType;
this.StartExecutorId = Throw.IfNullOrEmpty(startExecutorId);
if (outputType is not null && outputCollectorId is not null)
{
this.OutputType = outputType;
this.OutputCollectorId = outputCollectorId;
}
else if (outputCollectorId is not null)
{
throw new InvalidOperationException(
$"Either both or none of OutputType and OutputCollectorId must be set. ({nameof(outputType)}: {outputType} vs. {nameof(outputCollectorId)}: {outputCollectorId})"
);
}
this.OutputExecutorIds = outputExecutorIds ?? [];
}
public Dictionary<string, ExecutorInfo> Executors { get; }
public Dictionary<string, List<EdgeInfo>> Edges { get; }
public HashSet<InputPortInfo> InputPorts { get; }
public TypeId InputType { get; }
public TypeId? InputType { get; }
public string StartExecutorId { get; }
public TypeId? OutputType { get; }
public string? OutputCollectorId { get; }
public HashSet<string> OutputExecutorIds { get; }
private bool IsMatch(Workflow workflow)
public bool IsMatch(Workflow workflow)
{
if (workflow is null)
{
return false;
}
if (!this.InputType.IsMatch(workflow.InputType))
{
return false;
}
if (this.StartExecutorId != workflow.StartExecutorId)
{
return false;
@@ -101,13 +82,21 @@ internal sealed class WorkflowInfo
return false;
}
// Validate the outputs
if (workflow.OutputExecutors.Count != this.OutputExecutorIds.Count ||
this.OutputExecutorIds.Any(id => !workflow.OutputExecutors.Contains(id)))
{
return false;
}
return true;
}
public bool IsMatch<TInput>(Workflow<TInput> workflow) => this.IsMatch(workflow as Workflow);
public bool IsMatch<TInput>(Workflow<TInput> workflow) =>
this.IsMatch(workflow as Workflow) && this.InputType?.IsMatch<TInput>() == true;
public bool IsMatch<TInput, TResult>(Workflow<TInput, TResult> workflow)
=> this.IsMatch(workflow as Workflow)
&& this.OutputType?.IsMatch(typeof(TResult)) is true
&& this.OutputCollectorId is not null && this.OutputCollectorId == workflow.OutputCollectorId;
//public bool IsMatch<TInput, TResult>(WorkflowWithOutput<TInput, TResult> workflow)
// => this.IsMatch(workflow as Workflow)
// && this.OutputType?.IsMatch(typeof(TResult)) is true
// && this.OutputCollectorId is not null && this.OutputCollectorId == workflow.OutputCollectorId;
}
dotnet/src/Microsoft.Agents.Workflows/Execution/IRunnerWithOutput.cs
-10
View File
@@ -1,10 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
namespace Microsoft.Agents.Workflows.Execution;
internal interface IRunnerWithOutput<TResult>
{
ISuperStepRunner StepRunner { get; }
TResult? RunningOutput { get; }
}
dotnet/src/Microsoft.Agents.Workflows/Execution/ISuperStepRunner.cs
+2
View File
@@ -8,6 +8,8 @@ namespace Microsoft.Agents.Workflows.Execution;
internal interface ISuperStepRunner
{
string RunId { get; }
bool HasUnservicedRequests { get; }
bool HasUnprocessedMessages { get; }
dotnet/src/Microsoft.Agents.Workflows/Execution/MessageRouter.cs
+27 -8
View File
@@ -2,11 +2,17 @@
using System;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.Agents.Workflows.Checkpointing;
using Microsoft.Shared.Diagnostics;
using CatchAllF =
System.Func<
Microsoft.Agents.Workflows.PortableValue, // message
Microsoft.Agents.Workflows.IWorkflowContext, // context
System.Threading.Tasks.ValueTask<Microsoft.Agents.Workflows.Execution.CallResult>
>;
using MessageHandlerF =
System.Func<
object, // message
@@ -20,36 +26,44 @@ internal sealed class MessageRouter
{
private readonly Dictionary<Type, MessageHandlerF> _typedHandlers;
private readonly Dictionary<TypeId, Type> _runtimeTypeMap;
private readonly MessageHandlerF? _catchAllHandler;
internal MessageRouter(Dictionary<Type, MessageHandlerF> handlers)
private readonly CatchAllF? _catchAllFunc;
internal MessageRouter(Dictionary<Type, MessageHandlerF> handlers, HashSet<Type> outputTypes, CatchAllF? catchAllFunc)
{
Throw.IfNull(handlers);
this._typedHandlers = handlers;
this._runtimeTypeMap = handlers.Keys.ToDictionary(t => new TypeId(t), t => t);
this._catchAllHandler = handlers.FirstOrDefault(e => e.Key == typeof(object)).Value;
this._catchAllFunc = catchAllFunc;
this.IncomingTypes = [.. handlers.Keys];
this.DefaultOutputTypes = outputTypes;
}
public HashSet<Type> IncomingTypes { get; }
[MemberNotNullWhen(true, nameof(_catchAllFunc))]
internal bool HasCatchAll => this._catchAllFunc is not null;
public bool CanHandle(object message) => this.CanHandle(new TypeId(Throw.IfNull(message).GetType()));
public bool CanHandle(Type candidateType) => this.CanHandle(new TypeId(Throw.IfNull(candidateType)));
public bool CanHandle(TypeId candidateType)
{
return this._catchAllHandler is not null || this._runtimeTypeMap.ContainsKey(candidateType);
return this.HasCatchAll || this._runtimeTypeMap.ContainsKey(candidateType);
}
public HashSet<Type> DefaultOutputTypes { get; }
public async ValueTask<CallResult?> RouteMessageAsync(object message, IWorkflowContext context, bool requireRoute = false)
{
Throw.IfNull(message);
CallResult? result = null;
if (message is PortableValue portableValue &&
PortableValue? portableValue = message as PortableValue;
if (portableValue != null &&
this._runtimeTypeMap.TryGetValue(portableValue.TypeId, out Type? runtimeType))
{
// If we found a runtime type, we can use it
@@ -58,11 +72,16 @@ internal sealed class MessageRouter
try
{
if (this._typedHandlers.TryGetValue(message.GetType(), out MessageHandlerF? handler) ||
(handler = this._catchAllHandler) is not null)
if (this._typedHandlers.TryGetValue(message.GetType(), out MessageHandlerF? handler))
{
result = await handler(message, context).ConfigureAwait(false);
}
else if (this.HasCatchAll)
{
portableValue ??= new PortableValue(message);
result = await this._catchAllFunc(portableValue, context).ConfigureAwait(false);
}
}
catch (Exception e)
{
dotnet/src/Microsoft.Agents.Workflows/Execution/OutputFilter.cs
+11
View File
@@ -0,0 +1,11 @@
// Copyright (c) Microsoft. All rights reserved.
namespace Microsoft.Agents.Workflows.Execution;
internal sealed class OutputFilter(Workflow workflow)
{
public bool CanOutput(string sourceExecutorId, object output)
{
return workflow.OutputExecutors.Contains(sourceExecutorId);
}
}
dotnet/src/Microsoft.Agents.Workflows/Executor.cs
+45 -8
View File
@@ -28,11 +28,11 @@ public abstract class Executor : IIdentified
/// <summary>
/// Initialize the executor with a unique identifier
/// </summary>
/// <param name="id">A optional unique identifier for the executor. If <c>null</c>, a type-tagged UUID will be generated.</param>
/// <param name="id">A unique identifier for the executor.</param>
/// <param name="options">Configuration options for the executor. If <c>null</c>, default options will be used.</param>
protected Executor(string? id = null, ExecutorOptions? options = null)
protected Executor(string id, ExecutorOptions? options = null)
{
this.Id = id ?? $"{this.GetType().Name}/{Guid.NewGuid():N}";
this.Id = id;
this._options = options ?? ExecutorOptions.Default;
}
@@ -41,6 +41,26 @@ public abstract class Executor : IIdentified
/// </summary>
protected abstract RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder);
/// <summary>
/// Override this method to declare the types of messages this executor can send.
/// </summary>
/// <returns></returns>
protected virtual ISet<Type> ConfigureSentTypes() => new HashSet<Type>([typeof(object)]);
/// <summary>
/// Override this method to declare the types of messages this executor can yield as workflow outputs.
/// </summary>
/// <returns></returns>
protected virtual ISet<Type> ConfigureYieldTypes()
{
if (this._options.AutoYieldOutputHandlerResultObject)
{
return this.Router.DefaultOutputTypes;
}
return new HashSet<Type>();
}
private MessageRouter? _router;
internal MessageRouter Router
{
@@ -106,6 +126,10 @@ public abstract class Executor : IIdentified
{
await context.SendMessageAsync(result.Result).ConfigureAwait(false);
}
if (result.Result is not null && this._options.AutoYieldOutputHandlerResultObject)
{
await context.YieldOutputAsync(result.Result).ConfigureAwait(false);
}
return result.Result;
}
@@ -134,7 +158,7 @@ public abstract class Executor : IIdentified
/// <summary>
/// A set of <see cref="Type"/>s, representing the messages this executor can produce as output.
/// </summary>
public virtual ISet<Type> OutputTypes { get; } = new HashSet<Type>([typeof(object)]);
public ISet<Type> OutputTypes { get; } = new HashSet<Type>([typeof(object)]);
/// <summary>
/// Checks if the executor can handle a specific message type.
@@ -144,15 +168,28 @@ public abstract class Executor : IIdentified
public bool CanHandle(Type messageType) => this.Router.CanHandle(messageType);
internal bool CanHandle(TypeId messageType) => this.Router.CanHandle(messageType);
internal bool CanOutput(Type messageType)
{
foreach (Type type in this.OutputTypes)
{
if (type.IsAssignableFrom(messageType))
{
return true;
}
}
return false;
}
}
/// <summary>
/// Provides a simple executor implementation that uses a single message handler function to process incoming messages.
/// </summary>
/// <typeparam name="TInput">The type of input message.</typeparam>
/// <param name="id">A optional unique identifier for the executor. If <c>null</c>, a type-tagged UUID will be generated.</param>
/// <param name="id">A unique identifier for the executor.</param>
/// <param name="options">Configuration options for the executor. If <c>null</c>, default options will be used.</param>
public abstract class Executor<TInput>(string? id = null, ExecutorOptions? options = null)
public abstract class Executor<TInput>(string id, ExecutorOptions? options = null)
: Executor(id, options), IMessageHandler<TInput>
{
/// <inheritdoc/>
@@ -168,9 +205,9 @@ public abstract class Executor<TInput>(string? id = null, ExecutorOptions? optio
/// </summary>
/// <typeparam name="TInput">The type of input message.</typeparam>
/// <typeparam name="TOutput">The type of output message.</typeparam>
/// <param name="id">A optional unique identifier for the executor. If <c>null</c>, a type-tagged UUID will be generated.</param>
/// <param name="id">A unique identifier for the executor.</param>
/// <param name="options">Configuration options for the executor. If <c>null</c>, default options will be used.</param>
public abstract class Executor<TInput, TOutput>(string? id = null, ExecutorOptions? options = null)
public abstract class Executor<TInput, TOutput>(string id, ExecutorOptions? options = null)
: Executor(id, options ?? ExecutorOptions.Default),
IMessageHandler<TInput, TOutput>
{
dotnet/src/Microsoft.Agents.Workflows/ExecutorIsh.cs
+17 -5
View File
@@ -20,7 +20,7 @@ public static class ExecutorIshConfigurationExtensions
/// </summary>
/// <remarks>
/// Although this will generally result in a delay-instantiated <see cref="Executor"/> once messages are available
/// for it, if this is used as a start node of a typed <see cref="Workflow{TInput}"/> via <see cref="WorkflowBuilder.Build{T}"/>,
/// for it, if this is used as a start node of a typed <see cref="Workflow{TInput}"/> via <see cref="Workflow.TryPromoteAsync{TInput}"/>,
/// it will be instantiated as part of the workflow's construction, to validate that its input type matches the
/// demanded <c>TInput</c>.
/// </remarks>
@@ -55,11 +55,11 @@ public static class ExecutorIshConfigurationExtensions
/// </summary>
/// <typeparam name="TInput">The type of input message.</typeparam>
/// <param name="messageHandlerAsync">A delegate that defines the asynchronous function to execute for each input message.</param>
/// <param name="id">A optional unique identifier for the executor. If <c>null</c>, a type-tagged UUID will be generated.</param>
/// <param name="id">A optional unique identifier for the executor. If <c>null</c>, will use the function argument as an id.</param>
/// <param name="options">Configuration options for the executor. If <c>null</c>, default options will be used.</param>
/// <returns>An ExecutorIsh instance that wraps the provided asynchronous message handler and configuration.</returns>
public static ExecutorIsh AsExecutor<TInput>(this Func<TInput, IWorkflowContext, CancellationToken, ValueTask> messageHandlerAsync, string id, ExecutorOptions? options = null)
=> new FunctionExecutor<TInput>(messageHandlerAsync, id, options).ToExecutorIsh(messageHandlerAsync);
=> new FunctionExecutor<TInput>(id, messageHandlerAsync, options).ToExecutorIsh(messageHandlerAsync);
/// <summary>
/// Configures a function-based asynchronous message handler as an executor with the specified identifier and
@@ -68,11 +68,23 @@ public static class ExecutorIshConfigurationExtensions
/// <typeparam name="TInput">The type of input message.</typeparam>
/// <typeparam name="TOutput">The type of output message.</typeparam>
/// <param name="messageHandlerAsync">A delegate that defines the asynchronous function to execute for each input message.</param>
/// <param name="id">A optional unique identifier for the executor. If <c>null</c>, a type-tagged UUID will be generated.</param>
/// <param name="id">A unique identifier for the executor.</param>
/// <param name="options">Configuration options for the executor. If <c>null</c>, default options will be used.</param>
/// <returns>An ExecutorIsh instance that wraps the provided asynchronous message handler and configuration.</returns>
public static ExecutorIsh AsExecutor<TInput, TOutput>(this Func<TInput, IWorkflowContext, CancellationToken, ValueTask<TOutput>> messageHandlerAsync, string id, ExecutorOptions? options = null)
=> new FunctionExecutor<TInput, TOutput>(messageHandlerAsync, id, options).ToExecutorIsh(messageHandlerAsync);
=> new FunctionExecutor<TInput, TOutput>(Throw.IfNull(id), messageHandlerAsync, options).ToExecutorIsh(messageHandlerAsync);
/// <summary>
/// Configures a function-based aggregating executor with the specified identifier and options.
/// </summary>
/// <typeparam name="TInput">The type of input message.</typeparam>
/// <typeparam name="TAccumulate">The type of the accumulating object.</typeparam>
/// <param name="aggregatorFunc">A delegate the defines the aggregation procedure</param>
/// <param name="id">A unique identifier for the executor.</param>
/// <param name="options">Configuration options for the executor. If <c>null</c>, default options will be used.</param>
/// <returns>An ExecutorIsh instance that wraps the provided asynchronous message handler and configuration.</returns>
public static ExecutorIsh AsExecutor<TInput, TAccumulate>(this Func<TAccumulate?, TInput, TAccumulate?> aggregatorFunc, string id, ExecutorOptions? options = null)
=> new AggregatingExecutor<TInput, TAccumulate>(id, aggregatorFunc, options);
}
/// <summary>
dotnet/src/Microsoft.Agents.Workflows/ExecutorOptions.cs
+6 -1
View File
@@ -15,7 +15,12 @@ public class ExecutorOptions
internal ExecutorOptions() { }
/// <summary>
/// If <see langword="true"/>, the result of a message handler that returns a value will be sent as a message to the workflow.
/// If <see langword="true"/>, the result of a message handler that returns a value will be sent as a message from the executor.
/// </summary>
public bool AutoSendMessageHandlerResultObject { get; set; } = true;
/// <summary>
/// If <see langword="true"/>, the result of a message handler that returns a value will be yielded as an output of the executor.
/// </summary>
public bool AutoYieldOutputHandlerResultObject { get; set; } = true;
}
dotnet/src/Microsoft.Agents.Workflows/ExternalRequest.cs
+5
View File
@@ -77,6 +77,11 @@ public record ExternalRequest(InputPortInfo PortInfo, string RequestId, Portable
return new ExternalResponse(this.PortInfo, this.RequestId, new PortableValue(data));
}
internal ExternalResponse RewrapResponse(ExternalResponse response)
{
return new ExternalResponse(this.PortInfo, this.RequestId, response.Data);
}
/// <summary>
/// Creates a new <see cref="ExternalResponse"/> corresponding to the request, with the speicified data payload.
/// </summary>
dotnet/src/Microsoft.Agents.Workflows/FunctionExecutor.cs
+10 -8
View File
@@ -10,11 +10,11 @@ namespace Microsoft.Agents.Workflows;
/// Executes a user-provided asynchronous function in response to workflow messages of the specified input type.
/// </summary>
/// <typeparam name="TInput">The type of input message.</typeparam>
/// <param name="id">A unique identifier for the executor.</param>
/// <param name="handlerAsync">A delegate that defines the asynchronous function to execute for each input message.</param>
/// <param name="id">A optional unique identifier for the executor. If <c>null</c>, a type-tagged UUID will be generated.</param>
/// <param name="options">Configuration options for the executor. If <c>null</c>, default options will be used.</param>
public class FunctionExecutor<TInput>(Func<TInput, IWorkflowContext, CancellationToken, ValueTask> handlerAsync,
string? id = null,
public class FunctionExecutor<TInput>(string id,
Func<TInput, IWorkflowContext, CancellationToken, ValueTask> handlerAsync,
ExecutorOptions? options = null) : Executor<TInput>(id, options)
{
internal static Func<TInput, IWorkflowContext, CancellationToken, ValueTask> WrapAction(Action<TInput, IWorkflowContext, CancellationToken> handlerSync)
@@ -34,8 +34,9 @@ public class FunctionExecutor<TInput>(Func<TInput, IWorkflowContext, Cancellatio
/// <summary>
/// Creates a new instance of the <see cref="FunctionExecutor{TInput}"/> class.
/// </summary>
/// <param name="id">A unique identifier for the executor.</param>
/// <param name="handlerSync">A synchronous function to execute for each input message and workflow context.</param>
public FunctionExecutor(Action<TInput, IWorkflowContext, CancellationToken> handlerSync) : this(WrapAction(handlerSync))
public FunctionExecutor(string id, Action<TInput, IWorkflowContext, CancellationToken> handlerSync) : this(id, WrapAction(handlerSync))
{
}
}
@@ -45,11 +46,11 @@ public class FunctionExecutor<TInput>(Func<TInput, IWorkflowContext, Cancellatio
/// </summary>
/// <typeparam name="TInput">The type of input message.</typeparam>
/// <typeparam name="TOutput">The type of output message.</typeparam>
/// <param name="id">A unique identifier for the executor.</param>
/// <param name="handlerAsync">A delegate that defines the asynchronous function to execute for each input message.</param>
/// <param name="id">A optional unique identifier for the executor. If <c>null</c>, a type-tagged UUID will be generated.</param>
/// <param name="options">Configuration options for the executor. If <c>null</c>, default options will be used.</param>
public class FunctionExecutor<TInput, TOutput>(Func<TInput, IWorkflowContext, CancellationToken, ValueTask<TOutput>> handlerAsync,
string? id = null,
public class FunctionExecutor<TInput, TOutput>(string id,
Func<TInput, IWorkflowContext, CancellationToken, ValueTask<TOutput>> handlerAsync,
ExecutorOptions? options = null) : Executor<TInput, TOutput>(id, options)
{
internal static Func<TInput, IWorkflowContext, CancellationToken, ValueTask<TOutput>> WrapFunc(Func<TInput, IWorkflowContext, CancellationToken, TOutput> handlerSync)
@@ -69,8 +70,9 @@ public class FunctionExecutor<TInput, TOutput>(Func<TInput, IWorkflowContext, Ca
/// <summary>
/// Creates a new instance of the <see cref="FunctionExecutor{TInput,TOutput}"/> class.
/// </summary>
/// <param name="id">A unique identifier for the executor.</param>
/// <param name="handlerSync">A synchronous function to execute for each input message and workflow context.</param>
public FunctionExecutor(Func<TInput, IWorkflowContext, CancellationToken, TOutput> handlerSync) : this(WrapFunc(handlerSync))
public FunctionExecutor(string id, Func<TInput, IWorkflowContext, CancellationToken, TOutput> handlerSync) : this(id, WrapFunc(handlerSync))
{
}
}
dotnet/src/Microsoft.Agents.Workflows/IWorkflowContext.cs
+18
View File
@@ -28,6 +28,24 @@ public interface IWorkflowContext
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
ValueTask SendMessageAsync(object message, string? targetId = null);
/// <summary>
/// Adds an output value to the workflow's output queue. These outputs will be bubbled out of the workflow using the
/// <see cref="WorkflowOutputEvent"/>
/// </summary>
/// <remarks>
/// The type of the output message must match one of the output types declared by the Executor. By default, the return
/// types of registered message handlers are considered output types, unless otherwise specified using <see cref="ExecutorOptions"/>.
/// </remarks>
/// <param name="output">The output value to be returned.</param>
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
ValueTask YieldOutputAsync(object output);
/// <summary>
/// Adds a request to "halt" workflow execution at the end of the current SuperStep.
/// </summary>
/// <returns></returns>
ValueTask RequestHaltAsync();
/// <summary>
/// Reads a state value from the workflow's state store. If no scope is provided, the executor's
/// default scope is used.
dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunner.cs
+61 -76
View File
@@ -16,47 +16,53 @@ namespace Microsoft.Agents.Workflows.InProc;
/// <summary>
/// Provides a local, in-process runner for executing a workflow using the specified input type.
/// </summary>
/// <remarks><para> <see cref="InProcessRunner{TInput}"/> enables step-by-step execution of a workflow graph entirely
/// <remarks><para> <see cref="InProcessRunner"/> enables step-by-step execution of a workflow graph entirely
/// within the current process, without distributed coordination. It is primarily intended for testing, debugging, or
/// scenarios where workflow execution does not require executor distribution. </para></remarks>
/// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
internal sealed class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointingRunner where TInput : notnull
internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingRunner
{
public InProcessRunner(Workflow<TInput> workflow, ICheckpointManager? checkpointManager, string? runId = null)
public InProcessRunner(Workflow workflow, ICheckpointManager? checkpointManager, string? runId = null, params Type[] knownValidInputTypes)
{
this.Workflow = Throw.IfNull(workflow);
this.RunContext = new InProcessRunnerContext<TInput>(workflow);
this.RunContext = new InProcessRunnerContext(workflow);
this.CheckpointManager = checkpointManager;
this.RunId = runId ?? Guid.NewGuid().ToString("N");
this._knownValidInputTypes = [.. knownValidInputTypes];
// Initialize the runners for each of the edges, along with the state for edges that
// need it.
this.EdgeMap = new EdgeMap(this.RunContext, this.Workflow.Edges, this.Workflow.Ports.Values, this.Workflow.StartExecutorId, this.StepTracer);
}
/// <inheritdoc cref="ISuperStepRunner.RunId"/>
public string RunId { get; }
public async ValueTask<bool> IsValidInputAsync<TMessage>(TMessage message)
private readonly HashSet<Type> _knownValidInputTypes;
public async ValueTask<bool> IsValidInputTypeAsync(Type messageType)
{
Throw.IfNull(message);
Type type = typeof(TMessage);
// Short circuit the logic if the type is the input type
if (type == typeof(TInput))
if (this._knownValidInputTypes.Contains(messageType))
{
return true;
}
Executor startingExecutor = await this.RunContext.EnsureExecutorAsync(this.Workflow.StartExecutorId, tracer: null).ConfigureAwait(false);
return startingExecutor.CanHandle(type);
if (startingExecutor.CanHandle(messageType))
{
this._knownValidInputTypes.Add(messageType);
return true;
}
return false;
}
async ValueTask<bool> ISuperStepRunner.EnqueueMessageAsync<T>(T message)
public async ValueTask<bool> EnqueueMessageAsync<T>(T message)
{
Throw.IfNull(message);
// Check that the type of the incoming message is compatible with the starting executor's
// input type.
if (!await this.IsValidInputAsync(message).ConfigureAwait(false))
if (!await this.IsValidInputTypeAsync(typeof(T)).ConfigureAwait(false))
{
return false;
}
@@ -65,14 +71,30 @@ internal sealed class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointing
return true;
}
public async ValueTask<bool> EnqueueMessageAsync(object message)
{
Throw.IfNull(message);
// Check that the type of the incoming message is compatible with the starting executor's
// input type.
if (!await this.IsValidInputTypeAsync(message.GetType()).ConfigureAwait(false))
{
return false;
}
await this.RunContext.AddExternalMessageUntypedAsync(message).ConfigureAwait(false);
return true;
}
ValueTask ISuperStepRunner.EnqueueResponseAsync(ExternalResponse response)
{
// TODO: Check that there exists a corresponding input port?
return this.RunContext.AddExternalMessageAsync(response);
}
private InProcStepTracer StepTracer { get; } = new();
private Workflow<TInput> Workflow { get; init; }
private InProcessRunnerContext<TInput> RunContext { get; init; }
private Workflow Workflow { get; init; }
private InProcessRunnerContext RunContext { get; init; }
private ICheckpointManager? CheckpointManager { get; }
private EdgeMap EdgeMap { get; init; }
@@ -122,9 +144,16 @@ internal sealed class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointing
return new StreamingRun(this);
}
public async ValueTask<StreamingRun> StreamAsync(TInput input, CancellationToken cancellation = default)
public async ValueTask<StreamingRun> StreamAsync(object input, CancellationToken cancellation = default)
{
await this.RunContext.AddExternalMessageAsync(input).ConfigureAwait(false);
await this.EnqueueMessageAsync(input).ConfigureAwait(false);
return new StreamingRun(this);
}
public async ValueTask<StreamingRun> StreamAsync<TInput>(TInput input, CancellationToken cancellation = default)
{
await this.EnqueueMessageAsync(input).ConfigureAwait(false);
return new StreamingRun(this);
}
@@ -137,7 +166,15 @@ internal sealed class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointing
return await Run.CaptureStreamAsync(streamingRun, cancellation).ConfigureAwait(false);
}
public async ValueTask<Run> RunAsync(TInput input, CancellationToken cancellation = default)
public async ValueTask<Run> RunAsync(object input, CancellationToken cancellation = default)
{
StreamingRun streamingRun = await this.StreamAsync(input, cancellation).ConfigureAwait(false);
cancellation.ThrowIfCancellationRequested();
return await Run.CaptureStreamAsync(streamingRun, cancellation).ConfigureAwait(false);
}
public async ValueTask<Run> RunAsync<TInput>(TInput input, CancellationToken cancellation = default)
{
StreamingRun streamingRun = await this.StreamAsync(input, cancellation).ConfigureAwait(false);
cancellation.ThrowIfCancellationRequested();
@@ -152,7 +189,10 @@ internal sealed class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointing
async ValueTask<bool> ISuperStepRunner.RunSuperStepAsync(CancellationToken cancellation)
{
cancellation.ThrowIfCancellationRequested();
if (cancellation.IsCancellationRequested)
{
return false;
}
StepContext currentStep = this.RunContext.Advance();
@@ -277,58 +317,3 @@ internal sealed class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointing
private bool CheckWorkflowMatch(Checkpoint checkpoint) =>
checkpoint.Workflow.IsMatch(this.Workflow);
}
internal sealed class InProcessRunner<TInput, TResult> : IRunnerWithOutput<TResult>, ICheckpointingRunner where TInput : notnull
{
private readonly Workflow<TInput, TResult> _workflow;
private readonly InProcessRunner<TInput> _innerRunner;
public InProcessRunner(Workflow<TInput, TResult> workflow, CheckpointManager? checkpointManager, string? runId = null)
{
this._workflow = Throw.IfNull(workflow);
this._innerRunner = new(workflow, checkpointManager, runId);
}
internal async ValueTask<StreamingRun<TResult>> ResumeStreamAsync(CheckpointInfo checkpoint, CancellationToken cancellation = default)
{
await this._innerRunner.ResumeStreamAsync(checkpoint, cancellation).ConfigureAwait(false);
return new StreamingRun<TResult>(this);
}
public async ValueTask<StreamingRun<TResult>> StreamAsync(TInput input, CancellationToken cancellation = default)
{
await ((ISuperStepRunner)this._innerRunner).EnqueueMessageAsync(input).ConfigureAwait(false);
return new StreamingRun<TResult>(this);
}
public async ValueTask<Run<TResult>> ResumeAsync(CheckpointInfo checkpoint, CancellationToken cancellation = default)
{
StreamingRun<TResult> streamingRun = await this.ResumeStreamAsync(checkpoint, cancellation).ConfigureAwait(false);
cancellation.ThrowIfCancellationRequested();
return await Run<TResult>.CaptureStreamAsync(streamingRun, cancellation).ConfigureAwait(false);
}
public async ValueTask<Run<TResult>> RunAsync(TInput input, CancellationToken cancellation = default)
{
StreamingRun<TResult> streamingRun = await this.StreamAsync(input, cancellation).ConfigureAwait(false);
cancellation.ThrowIfCancellationRequested();
return await Run<TResult>.CaptureStreamAsync(streamingRun, cancellation).ConfigureAwait(false);
}
public ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellation = default)
=> this._innerRunner.RestoreCheckpointAsync(checkpointInfo, cancellation);
internal ValueTask CheckpointAsync() => this._innerRunner.CheckpointAsync();
/// <inheritdoc cref="Workflow{TInput, TResult}.RunningOutput"/>
public TResult? RunningOutput => this._workflow.RunningOutput;
ISuperStepRunner IRunnerWithOutput<TResult>.StepRunner => this._innerRunner;
public IReadOnlyList<CheckpointInfo> Checkpoints => this._innerRunner.Checkpoints;
}
dotnet/src/Microsoft.Agents.Workflows/InProc/InProcessRunnerContext.cs
+23 -3
View File
@@ -14,16 +14,18 @@ using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Workflows.InProc;
internal sealed class InProcessRunnerContext<TExternalInput> : IRunnerContext
internal sealed class InProcessRunnerContext : IRunnerContext
{
private StepContext _nextStep = new();
private readonly Dictionary<string, ExecutorRegistration> _executorRegistrations;
private readonly Dictionary<string, Executor> _executors = [];
private readonly Dictionary<string, ExternalRequest> _externalRequests = [];
private readonly OutputFilter _outputFilter;
public InProcessRunnerContext(Workflow workflow, ILogger? logger = null)
{
this._executorRegistrations = Throw.IfNull(workflow).Registrations;
this._outputFilter = new(workflow);
}
public async ValueTask<Executor> EnsureExecutorAsync(string executorId, IStepTracer? tracer)
@@ -80,7 +82,7 @@ internal sealed class InProcessRunnerContext<TExternalInput> : IRunnerContext
return default;
}
public IWorkflowContext Bind(string executorId) => new BoundContext(this, executorId);
public IWorkflowContext Bind(string executorId) => new BoundContext(this, executorId, this._outputFilter);
public ValueTask PostAsync(ExternalRequest request)
{
@@ -94,11 +96,29 @@ internal sealed class InProcessRunnerContext<TExternalInput> : IRunnerContext
internal StateManager StateManager { get; } = new();
private sealed class BoundContext(InProcessRunnerContext<TExternalInput> RunnerContext, string ExecutorId) : IWorkflowContext
private sealed class BoundContext(InProcessRunnerContext RunnerContext, string ExecutorId, OutputFilter outputFilter) : IWorkflowContext
{
public ValueTask AddEventAsync(WorkflowEvent workflowEvent) => RunnerContext.AddEventAsync(workflowEvent);
public ValueTask SendMessageAsync(object message, string? targetId = null) => RunnerContext.SendMessageAsync(ExecutorId, message, targetId);
public async ValueTask YieldOutputAsync(object output)
{
Throw.IfNull(output);
Executor sourceExecutor = await RunnerContext.EnsureExecutorAsync(ExecutorId, tracer: null).ConfigureAwait(false);
if (!sourceExecutor.CanOutput(output.GetType()))
{
throw new InvalidOperationException($"Cannot output object of type {output.GetType().Name}. Expecting one of [{string.Join(", ", sourceExecutor.OutputTypes)}].");
}
if (outputFilter.CanOutput(ExecutorId, output))
{
await this.AddEventAsync(new WorkflowOutputEvent(output, ExecutorId)).ConfigureAwait(false);
}
}
public ValueTask RequestHaltAsync() => this.AddEventAsync(new RequestHaltEvent());
public ValueTask<T?> ReadStateAsync<T>(string key, string? scopeName = null)
=> RunnerContext.StateManager.ReadStateAsync<T>(ExecutorId, scopeName, key);
dotnet/src/Microsoft.Agents.Workflows/InProcessExecution.cs
+220 -168
View File
@@ -12,25 +12,115 @@ namespace Microsoft.Agents.Workflows;
/// </summary>
public static class InProcessExecution
{
internal static InProcessRunner CreateRunner(Workflow workflow, CheckpointManager? checkpointManager, string? runId)
=> new(workflow, checkpointManager, runId);
internal static InProcessRunner CreateRunner<TInput>(Workflow<TInput> checkedWorkflow, CheckpointManager? checkpointManager, string? runId)
where TInput : notnull
=> new(checkedWorkflow, checkpointManager, runId, [typeof(TInput)]);
private static ValueTask<StreamingRun> StreamAsync<TInput>(InProcessRunner runner, TInput input, CancellationToken cancellation = default)
=> runner.StreamAsync(input, cancellation);
private static ValueTask<StreamingRun> StreamAsync(InProcessRunner runner, object input, CancellationToken cancellation = default)
=> runner.StreamAsync(input, cancellation);
private static ValueTask<Run> RunAsync<TInput>(InProcessRunner runner, TInput input, CancellationToken cancellation = default)
=> runner.RunAsync(input, cancellation);
private static ValueTask<Run> RunAsync(InProcessRunner runner, object input, CancellationToken cancellation = default)
=> runner.RunAsync(input, cancellation);
private static async ValueTask<Checkpointed<StreamingRun>> StreamCheckpointedAsync<TInput>(InProcessRunner runner, TInput input, CancellationToken cancellation = default)
where TInput : notnull
{
StreamingRun run = await StreamAsync(runner, input, cancellation).ConfigureAwait(false);
await runner.CheckpointAsync(cancellation).ConfigureAwait(false);
return new(run, runner);
}
private static async ValueTask<Checkpointed<StreamingRun>> StreamCheckpointedAsync(InProcessRunner runner, object input, CancellationToken cancellation = default)
{
StreamingRun run = await StreamAsync(runner, input, cancellation).ConfigureAwait(false);
await runner.CheckpointAsync(cancellation).ConfigureAwait(false);
return new(run, runner);
}
private static async ValueTask<Checkpointed<Run>> RunCheckpointedAsync<TInput>(InProcessRunner runner, TInput input, CancellationToken cancellation = default)
where TInput : notnull
{
Run run = await RunAsync(runner, input, cancellation).ConfigureAwait(false);
await runner.CheckpointAsync(cancellation).ConfigureAwait(false);
return new(run, runner);
}
private static async ValueTask<Checkpointed<Run>> RunCheckpointedAsync(InProcessRunner runner, object input, CancellationToken cancellation = default)
{
Run run = await RunAsync(runner, input, cancellation).ConfigureAwait(false);
await runner.CheckpointAsync(cancellation).ConfigureAwait(false);
return new(run, runner);
}
private static async ValueTask<Checkpointed<StreamingRun>> ResumeStreamCheckpointedAsync(InProcessRunner runner, CheckpointInfo fromCheckpoint, CancellationToken cancellation = default)
{
StreamingRun run = await runner.ResumeStreamAsync(fromCheckpoint, cancellation).ConfigureAwait(false);
return new(run, runner);
}
private static async ValueTask<Checkpointed<Run>> ResumeRunCheckpointedAsync(InProcessRunner runner, CheckpointInfo fromCheckpoint, CancellationToken cancellation = default)
{
Run run = await runner.ResumeAsync(fromCheckpoint, cancellation).ConfigureAwait(false);
return new(run, runner);
}
/// <summary>
/// Initiates an asynchronous streaming execution using the specified input.
/// </summary>
/// <remarks>The returned <see cref="StreamingRun"/> provides methods to observe and control
/// the ongoing streaming execution. The operation will continue until the streaming execution is finished or
/// cancelled.</remarks>
/// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
/// <typeparam name="TInput">A type of input accepted by the workflow. Must be non-nullable.</typeparam>
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="input">The input message to be processed as part of the streaming run.</param>
/// <param name="runId">An optional unique identifier for the run. If not provided, a new identifier will be generated.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="ValueTask{StreamingRun}"/> that represents the asynchronous operation. The result contains a <see
/// cref="StreamingRun"/> for managing and interacting with the streaming run.</returns>
public static ValueTask<StreamingRun> StreamAsync<TInput>(
Workflow workflow,
TInput input,
string? runId = null,
CancellationToken cancellation = default) where TInput : notnull
{
InProcessRunner runner = CreateRunner(workflow, checkpointManager: null, runId);
return StreamAsync(runner, (object)input, cancellation);
}
/// <summary>
/// Initiates an asynchronous streaming execution using the specified input.
/// </summary>
/// <remarks>The returned <see cref="StreamingRun"/> provides methods to observe and control
/// the ongoing streaming execution. The operation will continue until the streaming execution is finished or
/// cancelled.</remarks>
/// <typeparam name="TInput">A type of input accepted by the workflow. Must be non-nullable.</typeparam>
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="input">The input message to be processed as part of the streaming run.</param>
/// <param name="runId">An optional unique identifier for the run. If not provided, a new identifier will be generated.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="ValueTask{StreamingRun}"/> that represents the asynchronous operation. The result contains a <see
/// cref="StreamingRun"/> for managing and interacting with the streaming run.</returns>
public static ValueTask<StreamingRun> StreamAsync<TInput>(
Workflow<TInput> workflow,
TInput input,
string? runId = null,
CancellationToken cancellation = default) where TInput : notnull
{
InProcessRunner<TInput> runner = new(workflow, checkpointManager: null);
return runner.StreamAsync(input, cancellation);
InProcessRunner runner = CreateRunner(workflow, checkpointManager: null, runId);
return StreamAsync(runner, input, cancellation);
}
/// <summary>
@@ -43,122 +133,89 @@ public static class InProcessExecution
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="input">The input message to be processed as part of the streaming run.</param>
/// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
/// <param name="runId">An optional unique identifier for the run. If not provided, a new identifier will be generated.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="ValueTask{StreamingRun}"/> that represents the asynchronous operation. The result contains a <see
/// cref="StreamingRun"/> for managing and interacting with the streaming run.</returns>
public static async ValueTask<Checkpointed<StreamingRun>> StreamAsync<TInput>(
public static ValueTask<Checkpointed<StreamingRun>> StreamAsync<TInput>(
Workflow workflow,
TInput input,
CheckpointManager checkpointManager,
string? runId = null,
CancellationToken cancellation = default) where TInput : notnull
{
InProcessRunner runner = CreateRunner(workflow, checkpointManager, runId);
return StreamCheckpointedAsync(runner, (object)input, cancellation);
}
/// <summary>
/// Initiates an asynchronous streaming execution using the specified input, with checkpointing.
/// </summary>
/// <remarks>The returned <see cref="StreamingRun"/> provides methods to observe and control
/// the ongoing streaming execution. The operation will continue until the streaming execution is finished or
/// cancelled.</remarks>
/// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="input">The input message to be processed as part of the streaming run.</param>
/// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
/// <param name="runId">An optional unique identifier for the run. If not provided, a new identifier will be generated.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="ValueTask{StreamingRun}"/> that represents the asynchronous operation. The result contains a <see
/// cref="StreamingRun"/> for managing and interacting with the streaming run.</returns>
public static ValueTask<Checkpointed<StreamingRun>> StreamAsync<TInput>(
Workflow<TInput> workflow,
TInput input,
CheckpointManager checkpointManager,
string? runId = null,
CancellationToken cancellation = default) where TInput : notnull
{
InProcessRunner<TInput> runner = new(workflow, checkpointManager);
StreamingRun result = await runner.StreamAsync(input, cancellation).ConfigureAwait(false);
await runner.CheckpointAsync(cancellation).ConfigureAwait(false);
return new(result, runner);
InProcessRunner runner = CreateRunner(workflow, checkpointManager, runId);
return StreamCheckpointedAsync(runner, input, cancellation);
}
/// <summary>
/// Resumes an asynchronous streaming execution for the specified input from a checkpoint.
/// </summary>
/// <remarks>The returned <see cref="StreamingRun{TResult}"/> can be used to retrieve results
/// as they become available. If the operation is cancelled via the <paramref name="cancellation"/> token, the
/// streaming execution will be terminated.</remarks>
/// <remarks>If the operation is cancelled via the <paramref name="cancellation"/> token, the streaming execution will
/// be terminated.</remarks>
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="fromCheckpoint">The <see cref="CheckpointInfo"/> corresponding to the checkpoint from which to resume.</param>
/// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
/// <param name="runId">An optional unique identifier for the run. If not provided, a new identifier will be generated.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="StreamingRun"/> that provides access to the results of the streaming run.</returns>
public static ValueTask<Checkpointed<StreamingRun>> ResumeStreamAsync(
Workflow workflow,
CheckpointInfo fromCheckpoint,
CheckpointManager checkpointManager,
string? runId = null,
CancellationToken cancellation = default)
{
InProcessRunner runner = CreateRunner(workflow, checkpointManager, runId);
return ResumeStreamCheckpointedAsync(runner, fromCheckpoint, cancellation);
}
/// <summary>
/// Resumes an asynchronous streaming execution for the specified input from a checkpoint.
/// </summary>
/// <remarks>If the operation is cancelled via the <paramref name="cancellation"/> token, the streaming execution will
/// be terminated.</remarks>
/// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="fromCheckpoint">The <see cref="CheckpointInfo"/> corresponding to the checkpoint from which to resume.</param>
/// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
/// <param name="runId">An optional unique identifier for the run. If not provided, a new identifier will be generated.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="StreamingRun{TResult}"/> that provides access to the results of the streaming
/// run.</returns>
public static async ValueTask<Checkpointed<StreamingRun>> ResumeStreamAsync<TInput>(
/// <returns>A <see cref="StreamingRun"/> that provides access to the results of the streaming run.</returns>
public static ValueTask<Checkpointed<StreamingRun>> ResumeStreamAsync<TInput>(
Workflow<TInput> workflow,
CheckpointInfo fromCheckpoint,
CheckpointManager checkpointManager,
string? runId = null,
CancellationToken cancellation = default) where TInput : notnull
{
InProcessRunner<TInput> runner = new(workflow, checkpointManager, runId: fromCheckpoint.RunId);
StreamingRun result = await runner.ResumeStreamAsync(fromCheckpoint, cancellation).ConfigureAwait(false);
return new(result, runner);
}
/// <summary>
/// Initiates an asynchronous streaming execution for the specified input.
/// </summary>
/// <remarks>The returned <see cref="StreamingRun{TResult}"/> can be used to retrieve results
/// as they become available. If the operation is cancelled via the <paramref name="cancellation"/> token, the
/// streaming execution will be terminated.</remarks>
/// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
/// <typeparam name="TResult">The type of output produced by the workflow.</typeparam>
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="input">The input value to be processed by the streaming run.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="StreamingRun{TResult}"/> that provides access to the results of the streaming
/// run.</returns>
public static ValueTask<StreamingRun<TResult>> StreamAsync<TInput, TResult>(
Workflow<TInput, TResult> workflow,
TInput input,
CancellationToken cancellation = default) where TInput : notnull
{
InProcessRunner<TInput, TResult> runner = new(workflow, checkpointManager: null);
return runner.StreamAsync(input, cancellation);
}
/// <summary>
/// Initiates an asynchronous streaming execution for the specified input, with checkpointing.
/// </summary>
/// <remarks>The returned <see cref="StreamingRun{TResult}"/> can be used to retrieve results
/// as they become available. If the operation is cancelled via the <paramref name="cancellation"/> token, the
/// streaming execution will be terminated.</remarks>
/// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
/// <typeparam name="TResult">The type of output produced by the workflow.</typeparam>
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="input">The input value to be processed by the streaming run.</param>
/// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="StreamingRun{TResult}"/> that provides access to the results of the streaming
/// run.</returns>
public static async ValueTask<Checkpointed<StreamingRun<TResult>>> StreamAsync<TInput, TResult>(
Workflow<TInput, TResult> workflow,
TInput input,
CheckpointManager checkpointManager,
CancellationToken cancellation = default) where TInput : notnull
{
InProcessRunner<TInput, TResult> runner = new(workflow, checkpointManager);
StreamingRun<TResult> result = await runner.StreamAsync(input, cancellation).ConfigureAwait(false);
await runner.CheckpointAsync().ConfigureAwait(false);
return new(result, runner);
}
/// <summary>
/// Resumes an asynchronous streaming execution of the workflow from a checkpoint.
/// </summary>
/// <remarks>The returned <see cref="StreamingRun{TResult}"/> can be used to retrieve results
/// as they become available. If the operation is cancelled via the <paramref name="cancellation"/> token, the
/// streaming execution will be terminated.</remarks>
/// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
/// <typeparam name="TResult">The type of output produced by the workflow.</typeparam>
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="fromCheckpoint">The <see cref="CheckpointInfo"/> corresponding to the checkpoint from which to resume.</param>
/// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="StreamingRun{TResult}"/> that provides access to the results of the streaming
/// run.</returns>
public static async ValueTask<Checkpointed<StreamingRun<TResult>>> ResumeStreamAsync<TInput, TResult>(
Workflow<TInput, TResult> workflow,
CheckpointInfo fromCheckpoint,
CheckpointManager checkpointManager,
CancellationToken cancellation = default) where TInput : notnull
{
InProcessRunner<TInput, TResult> runner = new(workflow, checkpointManager, runId: fromCheckpoint.RunId);
StreamingRun<TResult> result = await runner.ResumeStreamAsync(fromCheckpoint, cancellation).ConfigureAwait(false);
return new(result, runner);
InProcessRunner runner = CreateRunner(workflow, checkpointManager, runId);
return ResumeStreamCheckpointedAsync(runner, fromCheckpoint, cancellation);
}
/// <summary>
@@ -169,16 +226,40 @@ public static class InProcessExecution
/// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="input">The input message to be processed as part of the run.</param>
/// <param name="runId">An optional unique identifier for the run. If not provided, a new identifier will be generated.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="ValueTask{Run}"/> that represents the asynchronous operation. The result contains a <see
/// cref="Run"/> for managing and interacting with the streaming run.</returns>
public static ValueTask<Run> RunAsync<TInput>(
Workflow workflow,
TInput input,
string? runId = null,
CancellationToken cancellation = default) where TInput : notnull
{
InProcessRunner runner = CreateRunner(workflow, checkpointManager: null, runId);
return RunAsync(runner, (object)input, cancellation);
}
/// <summary>
/// Initiates a non-streaming execution of the workflow with the specified input.
/// </summary>
/// <remarks>The workflow will run until its first halt, and the returned <see cref="Run"/> will capture
/// all outgoing events. Use the <c>Run</c> instance to resume execution with responses to outgoing events.</remarks>
/// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="input">The input message to be processed as part of the run.</param>
/// <param name="runId">An optional unique identifier for the run. If not provided, a new identifier will be generated.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="ValueTask{Run}"/> that represents the asynchronous operation. The result contains a <see
/// cref="Run"/> for managing and interacting with the streaming run.</returns>
public static ValueTask<Run> RunAsync<TInput>(
Workflow<TInput> workflow,
TInput input,
string? runId = null,
CancellationToken cancellation = default) where TInput : notnull
{
InProcessRunner<TInput> runner = new(workflow, checkpointManager: null);
return runner.RunAsync(input, cancellation);
InProcessRunner runner = CreateRunner(workflow, checkpointManager: null, runId);
return RunAsync(runner, input, cancellation);
}
/// <summary>
@@ -190,66 +271,19 @@ public static class InProcessExecution
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="input">The input message to be processed as part of the run.</param>
/// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
/// <param name="runId">An optional unique identifier for the run. If not provided, a new identifier will be generated.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="ValueTask{Run}"/> that represents the asynchronous operation. The result contains a <see
/// cref="Run"/> for managing and interacting with the streaming run.</returns>
public static async ValueTask<Checkpointed<Run>> RunAsync<TInput>(
Workflow<TInput> workflow,
public static ValueTask<Checkpointed<Run>> RunAsync<TInput>(
Workflow workflow,
TInput input,
CheckpointManager checkpointManager,
string? runId = null,
CancellationToken cancellation = default) where TInput : notnull
{
InProcessRunner<TInput> runner = new(workflow, checkpointManager);
Run result = await runner.RunAsync(input, cancellation).ConfigureAwait(false);
await runner.CheckpointAsync(cancellation).ConfigureAwait(false);
return new(result, runner);
}
/// <summary>
/// Resumes a non-streaming execution of the workflow from a checkpoint.
/// </summary>
/// <remarks>The workflow will run until its first halt, and the returned <see cref="Run"/> will capture
/// all outgoing events. Use the <c>Run</c> instance to resume execution with responses to outgoing events.</remarks>
/// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="fromCheckpoint">The <see cref="CheckpointInfo"/> corresponding to the checkpoint from which to resume.</param>
/// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="ValueTask{Run}"/> that represents the asynchronous operation. The result contains a <see
/// cref="Run"/> for managing and interacting with the streaming run.</returns>
public static async ValueTask<Checkpointed<Run>> ResumeAsync<TInput>(
Workflow<TInput> workflow,
CheckpointInfo fromCheckpoint,
CheckpointManager checkpointManager,
CancellationToken cancellation = default) where TInput : notnull
{
InProcessRunner<TInput> runner = new(workflow, checkpointManager, runId: fromCheckpoint.RunId);
Run result = await runner.ResumeAsync(fromCheckpoint, cancellation).ConfigureAwait(false);
return new(result, runner);
}
/// <summary>
/// Initiates a non-streaming execution of the workflow with the specified input.
/// </summary>
/// <remarks>The workflow will run until its first halt, and the returned <see cref="Run"/> will capture
/// all outgoing events. Use the <c>Run</c> instance to resume execution with responses to outgoing events.</remarks>
/// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
/// <typeparam name="TResult">The type of output produced by the workflow.</typeparam>
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="input">The input message to be processed as part of the run.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="ValueTask{Run}"/> that represents the asynchronous operation. The result contains a <see
/// cref="Run"/> for managing and interacting with the streaming run.</returns>
public static ValueTask<Run<TResult>> RunAsync<TInput, TResult>(
Workflow<TInput, TResult> workflow,
TInput input,
CancellationToken cancellation = default) where TInput : notnull
{
InProcessRunner<TInput, TResult> runner = new(workflow, checkpointManager: null);
return runner.RunAsync(input, cancellation);
InProcessRunner runner = CreateRunner(workflow, checkpointManager: checkpointManager, runId);
return RunCheckpointedAsync(runner, (object)input, cancellation);
}
/// <summary>
@@ -258,25 +292,22 @@ public static class InProcessExecution
/// <remarks>The workflow will run until its first halt, and the returned <see cref="Run"/> will capture
/// all outgoing events. Use the <c>Run</c> instance to resume execution with responses to outgoing events.</remarks>
/// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
/// <typeparam name="TResult">The type of output produced by the workflow.</typeparam>
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="input">The input message to be processed as part of the run.</param>
/// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
/// <param name="runId">An optional unique identifier for the run. If not provided, a new identifier will be generated.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="ValueTask{Run}"/> that represents the asynchronous operation. The result contains a <see
/// cref="Run"/> for managing and interacting with the streaming run.</returns>
public static async ValueTask<Checkpointed<Run<TResult>>> RunAsync<TInput, TResult>(
Workflow<TInput, TResult> workflow,
public static ValueTask<Checkpointed<Run>> RunAsync<TInput>(
Workflow<TInput> workflow,
TInput input,
CheckpointManager checkpointManager,
string? runId = null,
CancellationToken cancellation = default) where TInput : notnull
{
InProcessRunner<TInput, TResult> runner = new(workflow, checkpointManager);
Run<TResult> result = await runner.RunAsync(input, cancellation).ConfigureAwait(false);
await runner.CheckpointAsync().ConfigureAwait(false);
return new(result, runner);
InProcessRunner runner = CreateRunner(workflow, checkpointManager: checkpointManager, runId);
return RunCheckpointedAsync(runner, input, cancellation);
}
/// <summary>
@@ -284,23 +315,44 @@ public static class InProcessExecution
/// </summary>
/// <remarks>The workflow will run until its first halt, and the returned <see cref="Run"/> will capture
/// all outgoing events. Use the <c>Run</c> instance to resume execution with responses to outgoing events.</remarks>
/// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
/// <typeparam name="TResult">The type of output produced by the workflow.</typeparam>
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="fromCheckpoint">The <see cref="CheckpointInfo"/> corresponding to the checkpoint from which to resume.</param>
/// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
/// <param name="runId">An optional unique identifier for the run. If not provided, a new identifier will be generated.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="ValueTask{Run}"/> that represents the asynchronous operation. The result contains a <see
/// cref="Run"/> for managing and interacting with the streaming run.</returns>
public static async ValueTask<Checkpointed<Run<TResult>>> ResumeAsync<TInput, TResult>(
Workflow<TInput, TResult> workflow,
public static ValueTask<Checkpointed<Run>> ResumeAsync(
Workflow workflow,
CheckpointInfo fromCheckpoint,
CheckpointManager checkpointManager,
string? runId = null,
CancellationToken cancellation = default)
{
InProcessRunner runner = CreateRunner(workflow, checkpointManager, runId);
return ResumeRunCheckpointedAsync(runner, fromCheckpoint, cancellation);
}
/// <summary>
/// Resumes a non-streaming execution of the workflow from a checkpoint.
/// </summary>
/// <remarks>The workflow will run until its first halt, and the returned <see cref="Run"/> will capture
/// all outgoing events. Use the <c>Run</c> instance to resume execution with responses to outgoing events.</remarks>
/// <param name="workflow">The workflow to be executed. Must not be <c>null</c>.</param>
/// <param name="fromCheckpoint">The <see cref="CheckpointInfo"/> corresponding to the checkpoint from which to resume.</param>
/// <param name="checkpointManager">The <see cref="CheckpointManager"/> to use with this run.</param>
/// <param name="runId">An optional unique identifier for the run. If not provided, a new identifier will be generated.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation.</param>
/// <returns>A <see cref="ValueTask{Run}"/> that represents the asynchronous operation. The result contains a <see
/// cref="Run"/> for managing and interacting with the streaming run.</returns>
public static ValueTask<Checkpointed<Run>> ResumeAsync<TInput>(
Workflow<TInput> workflow,
CheckpointInfo fromCheckpoint,
CheckpointManager checkpointManager,
string? runId = null,
CancellationToken cancellation = default) where TInput : notnull
{
InProcessRunner<TInput, TResult> runner = new(workflow, checkpointManager, runId: fromCheckpoint.RunId);
Run<TResult> result = await runner.ResumeAsync(fromCheckpoint, cancellation).ConfigureAwait(false);
return new(result, runner);
InProcessRunner runner = CreateRunner(workflow, checkpointManager, runId);
return ResumeRunCheckpointedAsync(runner, fromCheckpoint, cancellation);
}
}
dotnet/src/Microsoft.Agents.Workflows/Reflection/ReflectingExecutor.cs
+2 -2
View File
@@ -16,8 +16,8 @@ public class ReflectingExecutor<
] TExecutor
> : Executor where TExecutor : ReflectingExecutor<TExecutor>
{
/// <inheritdoc cref="Executor(string?, ExecutorOptions?)"/>
protected ReflectingExecutor(string? id = null, ExecutorOptions? options = null) : base(id, options)
/// <inheritdoc cref="Executor(string, ExecutorOptions?)"/>
protected ReflectingExecutor(string id, ExecutorOptions? options = null) : base(id, options)
{
}
dotnet/src/Microsoft.Agents.Workflows/Reflection/RouteBuilderExtensions.cs
+1 -1
View File
@@ -91,7 +91,7 @@ internal static class RouteBuilderExtensions
foreach (MessageHandlerInfo handlerInfo in executorType.GetHandlerInfos())
{
builder = builder.AddHandler(handlerInfo.InType, handlerInfo.Bind(executor, checkType: true));
builder = builder.AddHandlerInternal(handlerInfo.InType, handlerInfo.Bind(executor, checkType: true), handlerInfo.OutType);
}
return builder;
dotnet/src/Microsoft.Agents.Workflows/RequestHaltEvent.cs
+12
View File
@@ -0,0 +1,12 @@
// Copyright (c) Microsoft. All rights reserved.
namespace Microsoft.Agents.Workflows;
/// <summary>
/// Event triggered when a workflow completes execution.
/// </summary>
internal sealed class RequestHaltEvent : WorkflowEvent
{
internal RequestHaltEvent(object? result = null) : base(result)
{ }
}
dotnet/src/Microsoft.Agents.Workflows/RouteBuilder.cs
+142 -9
View File
@@ -2,10 +2,16 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Threading.Tasks;
using Microsoft.Agents.Workflows.Execution;
using Microsoft.Shared.Diagnostics;
using CatchAllF =
System.Func<
Microsoft.Agents.Workflows.PortableValue, // message
Microsoft.Agents.Workflows.IWorkflowContext, // context
System.Threading.Tasks.ValueTask<Microsoft.Agents.Workflows.Execution.CallResult>
>;
using MessageHandlerF =
System.Func<
object, // message
@@ -24,16 +30,35 @@ namespace Microsoft.Agents.Workflows;
public class RouteBuilder
{
private readonly Dictionary<Type, MessageHandlerF> _typedHandlers = [];
private readonly Dictionary<Type, Type> _outputTypes = [];
private CatchAllF? _catchAll;
internal RouteBuilder AddHandler(Type messageType, MessageHandlerF handler, bool overwrite = false)
internal RouteBuilder AddHandlerInternal(Type messageType, MessageHandlerF handler, Type? outputType, bool overwrite = false)
{
Throw.IfNull(messageType);
Throw.IfNull(handler);
if (messageType == typeof(PortableValue))
{
throw new InvalidOperationException("Cannot register a handler for PortableValue. Use AddCatchAll() instead.");
}
Debug.Assert(typeof(CallResult) != outputType, "Must not double-wrap message handlers in the RouteBuilder. " +
"Use AddHandlerInternal() or do not wrap user-provided handler.");
// Overwrite must be false if the type is not registered. Overwrite must be true if the type is registered.
if (this._typedHandlers.ContainsKey(messageType) == overwrite)
{
this._typedHandlers[messageType] = handler;
if (outputType is not null)
{
this._outputTypes[messageType] = outputType;
}
else
{
this._outputTypes.Remove(messageType);
}
}
else if (overwrite)
{
@@ -52,7 +77,7 @@ public class RouteBuilder
{
Throw.IfNull(handler);
return this.AddHandler(type, WrappedHandlerAsync, overwrite);
return this.AddHandlerInternal(type, WrappedHandlerAsync, outputType: null, overwrite);
async ValueTask<CallResult> WrappedHandlerAsync(object msg, IWorkflowContext ctx)
{
@@ -65,7 +90,7 @@ public class RouteBuilder
{
Throw.IfNull(handler);
return this.AddHandler(type, WrappedHandlerAsync, overwrite);
return this.AddHandlerInternal(type, WrappedHandlerAsync, outputType: typeof(TResult), overwrite);
async ValueTask<CallResult> WrappedHandlerAsync(object msg, IWorkflowContext ctx)
{
@@ -91,7 +116,7 @@ public class RouteBuilder
{
Throw.IfNull(handler);
return this.AddHandler(typeof(TInput), WrappedHandlerAsync, overwrite);
return this.AddHandlerInternal(typeof(TInput), WrappedHandlerAsync, outputType: null, overwrite);
async ValueTask<CallResult> WrappedHandlerAsync(object msg, IWorkflowContext ctx)
{
@@ -117,7 +142,7 @@ public class RouteBuilder
{
Throw.IfNull(handler);
return this.AddHandler(typeof(TInput), WrappedHandlerAsync, overwrite);
return this.AddHandlerInternal(typeof(TInput), WrappedHandlerAsync, outputType: null, overwrite);
async ValueTask<CallResult> WrappedHandlerAsync(object msg, IWorkflowContext ctx)
{
@@ -143,7 +168,7 @@ public class RouteBuilder
{
Throw.IfNull(handler);
return this.AddHandler(typeof(TInput), WrappedHandlerAsync, overwrite);
return this.AddHandlerInternal(typeof(TInput), WrappedHandlerAsync, outputType: typeof(TResult), overwrite);
async ValueTask<CallResult> WrappedHandlerAsync(object msg, IWorkflowContext ctx)
{
@@ -169,7 +194,7 @@ public class RouteBuilder
{
Throw.IfNull(handler);
return this.AddHandler(typeof(TInput), WrappedHandlerAsync, overwrite);
return this.AddHandlerInternal(typeof(TInput), WrappedHandlerAsync, outputType: typeof(TResult), overwrite);
async ValueTask<CallResult> WrappedHandlerAsync(object msg, IWorkflowContext ctx)
{
@@ -178,5 +203,113 @@ public class RouteBuilder
}
}
internal MessageRouter Build() => new(this._typedHandlers);
private RouteBuilder AddCatchAll(CatchAllF handler, bool overwrite = false)
{
if (!overwrite && this._catchAll != null)
{
throw new InvalidOperationException("A catch-all is already registered (overwrite = false).");
}
this._catchAll = handler;
return this;
}
/// <summary>
/// Register a handler function as a catch-all handler: It will be used if not type-matching handler is registered.
/// </summary>
/// <remarks>If a catch-all handler for already exists, setting <paramref name="overwrite"/> to <see langword="true"/>
/// will replace the existing handler; otherwise, an exception may be thrown. The handler receives the input message
/// wrapped as <see cref="PortableValue"/> and workflow context, and returns a result asynchronously.</remarks>
/// <param name="handler">A function that processes messages wrapped as <see cref="PortableValue"/> within the
/// workflow context. The delegate is invoked for each incoming message not otherwise handled.</param>
/// <param name="overwrite"><see langword="true"/> to replace any existing handler for the input type; otherwise, <see langword="false"/> to
/// preserve existing handlers.</param>
/// <returns>The current <see cref="RouteBuilder"/> instance, enabling fluent configuration of workflow routes.</returns>
public RouteBuilder AddCatchAll(Func<PortableValue, IWorkflowContext, ValueTask> handler, bool overwrite = false)
{
Throw.IfNull(handler);
return this.AddCatchAll(WrappedHandlerAsync, overwrite);
async ValueTask<CallResult> WrappedHandlerAsync(PortableValue message, IWorkflowContext ctx)
{
await handler.Invoke(message, ctx).ConfigureAwait(false);
return CallResult.ReturnVoid();
}
}
/// <summary>
/// Register a handler function as a catch-all handler: It will be used if not type-matching handler is registered.
/// </summary>
/// <remarks>If a catch-all handler for already exists, setting <paramref name="overwrite"/> to <see langword="true"/>
/// will replace the existing handler; otherwise, an exception may be thrown. The handler receives the input message
/// wrapped as <see cref="PortableValue"/> and workflow context, and returns a result asynchronously.</remarks>
/// <param name="handler">A function that processes messages wrapped as <see cref="PortableValue"/> within the
/// workflow context and returns a <see cref="ValueTask{TResult}"/> representing the asynchronous result.</param>
/// <param name="overwrite"><see langword="true"/> to replace any existing handler for the input type; otherwise, <see langword="false"/> to
/// preserve existing handlers.</param>
/// <returns>The current <see cref="RouteBuilder"/> instance, enabling fluent configuration of workflow routes.</returns>
public RouteBuilder AddCatchAll<TResult>(Func<PortableValue, IWorkflowContext, ValueTask<TResult>> handler, bool overwrite = false)
{
Throw.IfNull(handler);
return this.AddCatchAll(WrappedHandlerAsync, overwrite);
async ValueTask<CallResult> WrappedHandlerAsync(PortableValue message, IWorkflowContext ctx)
{
TResult result = await handler.Invoke(message, ctx).ConfigureAwait(false);
return CallResult.ReturnResult(result);
}
}
/// <summary>
/// Register a handler function as a catch-all handler: It will be used if not type-matching handler is registered.
/// </summary>
/// <remarks>If a catch-all handler for already exists, setting <paramref name="overwrite"/> to <see langword="true"/>
/// will replace the existing handler; otherwise, an exception may be thrown. The handler receives the input message
/// wrapped as <see cref="PortableValue"/> and workflow context, and returns a result asynchronously.</remarks>
/// <param name="handler">A function that processes messages wrapped as <see cref="PortableValue"/> within the
/// workflow context. The delegate is invoked for each incoming message not otherwise handled.</param>
/// <param name="overwrite"><see langword="true"/> to replace any existing handler for the input type; otherwise, <see langword="false"/> to
/// preserve existing handlers.</param>
/// <returns>The current <see cref="RouteBuilder"/> instance, enabling fluent configuration of workflow routes.</returns>
public RouteBuilder AddCatchAll(Action<PortableValue, IWorkflowContext> handler, bool overwrite = false)
{
Throw.IfNull(handler);
return this.AddCatchAll(WrappedHandlerAsync, overwrite);
ValueTask<CallResult> WrappedHandlerAsync(PortableValue message, IWorkflowContext ctx)
{
handler.Invoke(message, ctx);
return new(CallResult.ReturnVoid());
}
}
/// <summary>
/// Register a handler function as a catch-all handler: It will be used if not type-matching handler is registered.
/// </summary>
/// <remarks>If a catch-all handler for already exists, setting <paramref name="overwrite"/> to <see langword="true"/>
/// will replace the existing handler; otherwise, an exception may be thrown. The handler receives the input message
/// wrapped as <see cref="PortableValue"/> and workflow context, and returns a result asynchronously.</remarks>
/// <param name="handler">A function that processes messages wrapped as <see cref="PortableValue"/> within the
/// workflow context and returns a <see cref="ValueTask{TResult}"/> representing the asynchronous result.</param>
/// <param name="overwrite"><see langword="true"/> to replace any existing handler for the input type; otherwise, <see langword="false"/> to
/// preserve existing handlers.</param>
/// <returns>The current <see cref="RouteBuilder"/> instance, enabling fluent configuration of workflow routes.</returns>
public RouteBuilder AddCatchAll<TResult>(Func<PortableValue, IWorkflowContext, TResult> handler, bool overwrite = false)
{
Throw.IfNull(handler);
return this.AddCatchAll(WrappedHandlerAsync, overwrite);
ValueTask<CallResult> WrappedHandlerAsync(PortableValue message, IWorkflowContext ctx)
{
TResult result = handler.Invoke(message, ctx);
return new(CallResult.ReturnResult(result));
}
}
internal MessageRouter Build() => new(this._typedHandlers, [.. this._outputTypes.Values], this._catchAll);
}
dotnet/src/Microsoft.Agents.Workflows/Run.cs
+12 -38
View File
@@ -13,7 +13,7 @@ namespace Microsoft.Agents.Workflows;
public enum RunStatus
{
/// <summary>
/// The run has halted, has no outstanding requets, but has not received a <see cref="WorkflowCompletedEvent"/>.
/// The run has halted, has no outstanding requets, but has not received a <see cref="RequestHaltEvent"/>.
/// </summary>
Idle,
@@ -22,10 +22,11 @@ public enum RunStatus
/// </summary>
PendingRequests,
/// <summary>
/// The run has halted after receiving a <see cref="WorkflowCompletedEvent"/>.
/// </summary>
Completed,
// TODO: Figure out if we want to have some way to have a true "converged" state
///// <summary>
///// The run has halted after converging.
///// </summary>
//Completed,
/// <summary>
/// The workflow is currently running, and may receive events or requests.
@@ -56,31 +57,28 @@ public class Run
internal async ValueTask<bool> RunToNextHaltAsync(CancellationToken cancellation = default)
{
bool hadEvents = false;
bool hadCompletion = false;
this.Status = RunStatus.Running;
await foreach (WorkflowEvent evt in this._streamingRun.WatchStreamAsync(blockOnPendingRequest: false, cancellation).ConfigureAwait(false))
{
hadEvents = true;
if (evt is WorkflowCompletedEvent)
{
hadCompletion = true;
}
this._eventSink.Add(evt);
}
// TODO: bookmark every halt for history visualization?
this.Status =
hadCompletion
? RunStatus.Completed
: this._streamingRun.HasUnservicedRequests
this._streamingRun.HasUnservicedRequests
? RunStatus.PendingRequests
: RunStatus.Idle;
return hadEvents;
}
/// <summary>
/// A unique identifier for the run. Can be provided at the start of the run, or auto-generated.
/// </summary>
public string RunId => this._streamingRun.RunId;
/// <summary>
/// Gets the current execution status of the workflow run.
/// </summary>
@@ -150,27 +148,3 @@ public class Run
return await this.RunToNextHaltAsync(cancellation).ConfigureAwait(false);
}
}
/// <summary>
/// Represents a workflow run that tracks execution status and emitted workflow events, supporting resumption
/// with responses to <see cref="RequestInfoEvent"/>, and retrieval of the running output of the workflow.
/// </summary>
/// <typeparam name="TResult">The type of the workflow output.</typeparam>
public sealed class Run<TResult> : Run
{
internal static async ValueTask<Run<TResult>> CaptureStreamAsync(StreamingRun<TResult> run, CancellationToken cancellation = default)
{
Run<TResult> result = new(run);
await result.RunToNextHaltAsync(cancellation).ConfigureAwait(false);
return result;
}
private readonly StreamingRun<TResult> _streamingRun;
private Run(StreamingRun<TResult> streamingRun) : base(streamingRun)
{
this._streamingRun = streamingRun;
}
/// <inheritdoc cref="StreamingRun{TOutput}.RunningOutput"/>
public TResult? RunningOutput => this._streamingRun.RunningOutput;
}
dotnet/src/Microsoft.Agents.Workflows/Specialized/AIAgentHostExecutor.cs
+8 -27
View File
@@ -9,11 +9,10 @@ using Microsoft.Extensions.AI.Agents;
namespace Microsoft.Agents.Workflows.Specialized;
internal sealed class AIAgentHostExecutor : Executor
internal sealed class AIAgentHostExecutor : ChatProtocolExecutor
{
private readonly bool _emitEvents;
private readonly AIAgent _agent;
private readonly List<ChatMessage> _pendingMessages = [];
private AgentThread? _thread;
public AIAgentHostExecutor(AIAgent agent, bool emitEvents = false) : base(id: agent.Id)
@@ -25,13 +24,7 @@ internal sealed class AIAgentHostExecutor : Executor
private AgentThread EnsureThread(IWorkflowContext context) =>
this._thread ??= this._agent.GetNewThread();
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) =>
routeBuilder.AddHandler<ChatMessage>((message, _) => this._pendingMessages.Add(message))
.AddHandler<List<ChatMessage>>((messages, _) => this._pendingMessages.AddRange(messages))
.AddHandler<TurnToken>(this.TakeTurnAsync);
private const string ThreadStateKey = nameof(_thread);
private const string PendingMessagesStateKey = nameof(_pendingMessages);
protected internal override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default)
{
Task threadTask = Task.CompletedTask;
@@ -41,14 +34,9 @@ internal sealed class AIAgentHostExecutor : Executor
threadTask = context.QueueStateUpdateAsync(ThreadStateKey, threadValue).AsTask();
}
Task messagesTask = Task.CompletedTask;
if (this._pendingMessages.Count > 0)
{
JsonElement messagesValue = this._pendingMessages.Serialize();
messagesTask = context.QueueStateUpdateAsync(PendingMessagesStateKey, messagesValue).AsTask();
}
Task baseTask = base.OnCheckpointingAsync(context, cancellation).AsTask();
await Task.WhenAll(threadTask, messagesTask).ConfigureAwait(false);
await Task.WhenAll(threadTask, baseTask).ConfigureAwait(false);
}
protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellation = default)
@@ -59,18 +47,13 @@ internal sealed class AIAgentHostExecutor : Executor
this._thread = this._agent.DeserializeThread(threadValue.Value);
}
JsonElement? messagesValue = await context.ReadStateAsync<JsonElement?>(PendingMessagesStateKey).ConfigureAwait(false);
if (messagesValue.HasValue)
{
List<ChatMessage> messages = messagesValue.Value.DeserializeMessages();
this._pendingMessages.AddRange(messages);
}
await base.OnCheckpointRestoredAsync(context, cancellation).ConfigureAwait(false);
}
public async ValueTask TakeTurnAsync(TurnToken token, IWorkflowContext context)
protected override async ValueTask TakeTurnAsync(List<ChatMessage> messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellation = default)
{
bool emitEvents = token.EmitEvents ?? this._emitEvents;
IAsyncEnumerable<AgentRunResponseUpdate> agentStream = this._agent.RunStreamingAsync(this._pendingMessages, this.EnsureThread(context));
emitEvents ??= this._emitEvents;
IAsyncEnumerable<AgentRunResponseUpdate> agentStream = this._agent.RunStreamingAsync(messages, this.EnsureThread(context), cancellationToken: cancellation);
List<AIContent> updates = [];
ChatMessage? currentStreamingMessage = null;
@@ -83,7 +66,7 @@ internal sealed class AIAgentHostExecutor : Executor
continue;
}
if (emitEvents)
if (emitEvents ?? this._emitEvents)
{
await context.AddEventAsync(new AgentRunUpdateEvent(this.Id, update)).ConfigureAwait(false);
}
@@ -110,8 +93,6 @@ internal sealed class AIAgentHostExecutor : Executor
}
await PublishCurrentMessageAsync().ConfigureAwait(false);
this._pendingMessages.Clear();
await context.SendMessageAsync(token).ConfigureAwait(false);
async ValueTask PublishCurrentMessageAsync()
{
dotnet/src/Microsoft.Agents.Workflows/Specialized/IOutputSink.cs
-8
View File
@@ -1,8 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
namespace Microsoft.Agents.Workflows.Specialized;
internal interface IOutputSink<TResult> : IIdentified
{
TResult? Result { get; }
}
dotnet/src/Microsoft.Agents.Workflows/Specialized/OutputCollectorExecutor.cs
-37
View File
@@ -1,37 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Threading.Tasks;
using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Workflows.Specialized;
internal sealed class OutputCollectorExecutor<TInput, TResult> : Executor, IOutputSink<TResult>
{
private readonly StreamingAggregator<TInput, TResult> _aggregator;
private readonly Func<TInput, TResult?, bool>? _completionCondition;
public TResult? Result { get; private set; }
public OutputCollectorExecutor(StreamingAggregator<TInput, TResult> aggregator, Func<TInput, TResult?, bool>? completionCondition = null, string? id = null) : base(id)
{
this._aggregator = Throw.IfNull(aggregator);
this._completionCondition = completionCondition;
}
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) =>
routeBuilder.AddHandler<TInput>(this.HandleAsync);
public ValueTask HandleAsync(TInput message, IWorkflowContext context)
{
this.Result = this._aggregator(message, this.Result);
if (this._completionCondition is not null &&
this._completionCondition!(message, this.Result))
{
return context.AddEventAsync(new WorkflowCompletedEvent(this.Result));
}
return default;
}
}
dotnet/src/Microsoft.Agents.Workflows/Specialized/RequestInfoExecutor.cs
+55 -3
View File
@@ -1,6 +1,8 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Threading.Tasks;
using Microsoft.Agents.Workflows.Execution;
using Microsoft.Shared.Diagnostics;
@@ -9,6 +11,7 @@ namespace Microsoft.Agents.Workflows.Specialized;
internal sealed class RequestInfoExecutor : Executor
{
private readonly Dictionary<string, ExternalRequest> _wrappedRequests = new();
private InputPort Port { get; }
private IExternalRequestSink? RequestSink { get; set; }
@@ -32,12 +35,12 @@ internal sealed class RequestInfoExecutor : Executor
routeBuilder = routeBuilder
// Handle incoming requests (as raw request payloads)
.AddHandler(this.Port.Request, this.HandleAsync)
.AddHandler(typeof(object), this.HandleAsync);
.AddCatchAll(this.HandleCatchAllAsync);
if (this._allowWrapped)
{
routeBuilder = routeBuilder
.AddHandler<ExternalRequest, ExternalRequest>((request, context) => this.HandleAsync(request.Data, context));
.AddHandler<ExternalRequest, ExternalRequest>(this.HandleAsync);
}
return routeBuilder
@@ -47,9 +50,50 @@ internal sealed class RequestInfoExecutor : Executor
internal void AttachRequestSink(IExternalRequestSink requestSink) => this.RequestSink = Throw.IfNull(requestSink);
public async ValueTask<ExternalRequest> HandleCatchAllAsync(PortableValue message, IWorkflowContext context)
{
Throw.IfNull(message);
object? maybeRequest = message.AsType(this.Port.Request);
if (maybeRequest != null)
{
Debug.Assert(this.Port.Request.IsAssignableFrom(maybeRequest.GetType()));
ExternalRequest request = ExternalRequest.Create(this.Port, maybeRequest!);
await this.RequestSink!.PostAsync(request).ConfigureAwait(false);
}
throw new InvalidOperationException($"Message type {message.TypeId} could not be interpreted as a value of Request Type {this.Port.Request}");
}
public async ValueTask<ExternalRequest> HandleAsync(ExternalRequest message, IWorkflowContext context)
{
Debug.Assert(this._allowWrapped);
Throw.IfNull(message);
if (!message.Data.IsType(this.Port.Request))
{
throw new InvalidOperationException($"Message type {message.Data.TypeId} could not be interpreted as a value of Request Type {this.Port.Request}");
}
if (!message.PortInfo.ResponseType.IsMatchPolymorphic(this.Port.Response))
{
throw new InvalidOperationException($"Response type {this.Port.Response} is not a valid response for original request, whose expected response is {message.PortInfo.ResponseType}");
}
ExternalRequest request = ExternalRequest.Create(this.Port, message);
this._wrappedRequests.Add(request.RequestId, message);
await this.RequestSink!.PostAsync(request).ConfigureAwait(false);
return request;
}
public async ValueTask<ExternalRequest> HandleAsync(object message, IWorkflowContext context)
{
Throw.IfNull(message);
Debug.Assert(this.Port.Request.IsAssignableFrom(message.GetType()));
ExternalRequest request = ExternalRequest.Create(this.Port, message);
await this.RequestSink!.PostAsync(request).ConfigureAwait(false);
@@ -66,7 +110,15 @@ internal sealed class RequestInfoExecutor : Executor
throw new InvalidOperationException(
$"Message type {message.Data.TypeId} is not assignable to the response type {this.Port.Response.Name} of input port {this.Port.Id}.");
await context.SendMessageAsync(message).ConfigureAwait(false);
if (this._allowWrapped && this._wrappedRequests.TryGetValue(message.RequestId, out ExternalRequest? originalRequest))
{
await context.SendMessageAsync(originalRequest.RewrapResponse(message)).ConfigureAwait(false);
}
else
{
await context.SendMessageAsync(message).ConfigureAwait(false);
}
await context.SendMessageAsync(data).ConfigureAwait(false);
return message;
dotnet/src/Microsoft.Agents.Workflows/StreamingAggregators.cs
+24 -45
View File
@@ -6,17 +6,6 @@ using System.Linq;
namespace Microsoft.Agents.Workflows;
/// <summary>
/// Represents a function that incrementally aggregates a sequence of input values, producing an updated result for each
/// input.
/// </summary>
/// <typeparam name="TInput">The type of the input value to be aggregated.</typeparam>
/// <typeparam name="TResult">The type of the aggregation result produced by the function.</typeparam>
/// <param name="input">The current input value to be incorporated into the aggregation.</param>
/// <param name="runningResult">The current aggregated result, or null if this is the first input.</param>
/// <returns>The updated aggregation result after processing the input value, or null if no result can be produced.</returns>
public delegate TResult? StreamingAggregator<in TInput, TResult>(TInput input, TResult? runningResult);
/// <summary>
/// Provides a set of streaming aggregation functions for processing sequences of input values in a stateful,
/// incremental manner.
@@ -32,25 +21,17 @@ public static class StreamingAggregators
/// once.</remarks>
/// <typeparam name="TInput">The type of the input elements to be aggregated.</typeparam>
/// <typeparam name="TResult">The type of the result produced by the conversion function.</typeparam>
/// <param name="conversion">A function that converts an input value of type <typeparamref name="TInput"/> to a result of type <typeparamref
/// name="TResult"/>. This function is applied to the first input received.</param>
/// <returns>A <see cref="StreamingAggregator{TInput, TResult}"/> that yields the converted result of the first input.</returns>
public static StreamingAggregator<TInput, TResult> First<TInput, TResult>(Func<TInput, TResult> conversion)
/// <param name="conversion">A function that converts an input value of type <typeparamref name="TInput"/> to a result
/// of type <typeparamref name="TResult"/>. This function is applied to the first input received.</param>
/// <returns>An aggregation function that yields the result of converting the first input using the specified function.</returns>
public static Func<TResult?, TInput, TResult?> First<TInput, TResult>(Func<TInput, TResult> conversion)
{
bool hasRun = false;
TResult? local = default;
return Aggregate;
TResult? Aggregate(TInput input, TResult? runningResult)
TResult? Aggregate(TResult? runningResult, TInput input)
{
if (!hasRun)
{
local = conversion(input);
hasRun = true;
}
return local;
runningResult ??= conversion(input);
return runningResult;
}
}
@@ -58,8 +39,8 @@ public static class StreamingAggregators
/// Creates a streaming aggregator that returns the first input element.
/// </summary>
/// <typeparam name="TInput">The type of the input elements to aggregate.</typeparam>
/// <returns>A <see cref="StreamingAggregator{TInput, TInput}"/> that yields the first input element.</returns>
public static StreamingAggregator<TInput, TInput> First<TInput>() => First<TInput, TInput>(input => input);
/// <returns>A an aggrgation function that yields the first input element.</returns>
public static Func<TInput?, TInput, TInput?> First<TInput>() => First<TInput, TInput?>(input => input);
/// <summary>
/// Creates a streaming aggregator that returns the result of applying the specified conversion to the most recent
@@ -68,17 +49,15 @@ public static class StreamingAggregators
/// <typeparam name="TInput">The type of the input elements to be aggregated.</typeparam>
/// <typeparam name="TResult">The type of the result produced by the conversion function.</typeparam>
/// <param name="conversion">A function that converts each input value to a result. Cannot be null.</param>
/// <returns>A streaming aggregator that yields the converted value of the last input received.</returns>
public static StreamingAggregator<TInput, TResult> Last<TInput, TResult>(Func<TInput, TResult> conversion)
/// <returns>A aggregator function that yields the result of converting the last input received using the specified
/// function.</returns>
public static Func<TResult?, TInput, TResult?> Last<TInput, TResult>(Func<TInput, TResult> conversion)
{
TResult? local = default;
return Aggregate;
TResult? Aggregate(TInput input, TResult? runningResult)
TResult? Aggregate(TResult? runningResult, TInput input)
{
local = conversion(input);
return local;
return conversion(input);
}
}
@@ -86,8 +65,8 @@ public static class StreamingAggregators
/// Creates a streaming aggregator that returns the last element in a sequence.
/// </summary>
/// <typeparam name="TInput">The type of elements in the input sequence.</typeparam>
/// <returns>A <see cref="StreamingAggregator{TInput, TInput}"/> that yields the last element of the sequence.</returns>
public static StreamingAggregator<TInput, TInput> Last<TInput>() => Last<TInput, TInput>(input => input);
/// <returns>An aggregator function that yields the last element of the input.</returns>
public static Func<TInput?, TInput, TInput?> Last<TInput>() => Last<TInput, TInput?>(input => input);
/// <summary>
/// Creates a streaming aggregator that produces the union of results by applying a conversion function to each
@@ -96,13 +75,13 @@ public static class StreamingAggregators
/// <typeparam name="TInput">The type of the input elements to be aggregated.</typeparam>
/// <typeparam name="TResult">The type of the result elements produced by the conversion function.</typeparam>
/// <param name="conversion">A function that converts each input element to a result element to be included in the union.</param>
/// <returns>A streaming aggregator that, for each input, returns an enumerable containing all result elements produced so
/// far.</returns>
public static StreamingAggregator<TInput, IEnumerable<TResult>> Union<TInput, TResult>(Func<TInput, TResult> conversion)
/// <returns>An aggregator function that, for each input, returns an enumerable containing the result of converting every
/// element produced so far.</returns>
public static Func<IEnumerable<TResult>?, TInput, IEnumerable<TResult>?> Union<TInput, TResult>(Func<TInput, TResult> conversion)
{
return Aggregate;
IEnumerable<TResult> Aggregate(TInput input, IEnumerable<TResult>? runningResult)
IEnumerable<TResult> Aggregate(IEnumerable<TResult>? runningResult, TInput input)
{
return runningResult is not null ? runningResult.Append(conversion(input)) : [conversion(input)];
}
@@ -114,13 +93,13 @@ public static class StreamingAggregators
/// <remarks>The resulting aggregator combines all input sequences into a single sequence containing
/// distinct elements. The order of elements in the output sequence is not guaranteed.</remarks>
/// <typeparam name="TInput">The type of the elements in the input sequences to be aggregated.</typeparam>
/// <returns>A StreamingAggregator that, when applied to multiple input sequences, returns an IEnumerable containing the
/// union of all elements from those sequences.</returns>
public static StreamingAggregator<TInput, IEnumerable<TInput>> Union<TInput>()
/// <returns>An aggregator function, that, when applied to multiple input sequences, returns an <see cref="IEnumerable{TInput}"/>
/// containing the union of all elements from those sequences.</returns>
public static Func<IEnumerable<TInput>?, TInput, IEnumerable<TInput>?> Union<TInput>()
{
return Aggregate;
static IEnumerable<TInput> Aggregate(TInput input, IEnumerable<TInput>? runningResult)
static IEnumerable<TInput> Aggregate(IEnumerable<TInput>? runningResult, TInput input)
{
return runningResult is not null ? runningResult.Append(input) : [input];
}
dotnet/src/Microsoft.Agents.Workflows/StreamingRun.cs
+11 -46
View File
@@ -31,6 +31,11 @@ public class StreamingRun
this._stepRunner = Throw.IfNull(stepRunner);
}
/// <summary>
/// A unique identifier for the run. Can be provided at the start of the run, or auto-generated.
/// </summary>
public string RunId => this._stepRunner.RunId;
/// <summary>
/// Asynchronously sends the specified response to the external system and signals completion of the current
/// response wait operation.
@@ -72,7 +77,7 @@ public class StreamingRun
/// Asynchronously streams workflow events as they occur during workflow execution.
/// </summary>
/// <remarks>This method yields <see cref="WorkflowEvent"/> instances in real time as the workflow
/// progresses. The stream completes when a <see cref="WorkflowCompletedEvent"/> is encountered. Events are
/// progresses. The stream completes when a <see cref="RequestHaltEvent"/> is encountered. Events are
/// delivered in the order they are raised.</remarks>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the streaming operation. If cancellation is
/// requested, the stream will end and no further events will be yielded.</param>
@@ -104,18 +109,20 @@ public class StreamingRun
bool hadCompletionEvent = false;
foreach (WorkflowEvent raisedEvent in Interlocked.Exchange(ref eventSink, []))
{
yield return raisedEvent;
if (cancellation.IsCancellationRequested)
{
yield break; // Exit if cancellation is requested
}
// TODO: Do we actually want to interpret this as a termination request?
if (raisedEvent is WorkflowCompletedEvent)
if (raisedEvent is RequestHaltEvent)
{
hadCompletionEvent = true;
}
else
{
yield return raisedEvent;
}
}
if (hadCompletionEvent)
@@ -152,25 +159,6 @@ public class StreamingRun
}
}
/// <summary>
/// A <see cref="Workflow"/> run instance supporting a streaming form of receiving workflow events, providing
/// a mechanism to send responses back to the workflow, and retrieving the result of workflow execution.
/// </summary>
/// <typeparam name="TResult">The type of the workflow output.</typeparam>
public class StreamingRun<TResult> : StreamingRun
{
private readonly IRunnerWithOutput<TResult> _resultSource;
internal StreamingRun(IRunnerWithOutput<TResult> runner)
: base(Throw.IfNull(runner.StepRunner))
{
this._resultSource = runner;
}
/// <inheritdoc cref="IRunnerWithOutput{TResult}.RunningOutput"/>
public TResult? RunningOutput => this._resultSource.RunningOutput;
}
/// <summary>
/// Provides extension methods for processing and executing workflows using streaming runs.
/// </summary>
@@ -202,27 +190,4 @@ public static class StreamingRunExtensions
}
}
}
/// <summary>
/// Executes the workflow associated with the specified <see cref="StreamingRun{TResult}"/> until it
/// completes and returns the final result.
/// </summary>
/// <remarks>This method ensures that the workflow runs to completion before returning the result. If an
/// <paramref name="eventCallback"/> is provided, it will be invoked for each event emitted during the workflow's
/// execution, allowing for custom event handling.</remarks>
/// <typeparam name="TResult">The type of the result produced by the workflow.</typeparam>
/// <param name="handle">The <see cref="StreamingRun{TResult}"/> representing the workflow to execute.</param>
/// <param name="eventCallback">An optional callback function that is invoked for each <see cref="WorkflowEvent"/>
/// emitted during execution. The callback can process the event and return an object, or <see langword="null"/>
/// if no response is required.</param>
/// <param name="cancellation">A <see cref="CancellationToken"/> that can be used to cancel the workflow execution.</param>
/// <returns>A <see cref="ValueTask{TResult}"/> that represents the asynchronous operation. The task's result is the final
/// result of the workflow execution.</returns>
public static async ValueTask<TResult> RunToCompletionAsync<TResult>(this StreamingRun<TResult> handle, Func<WorkflowEvent, object?>? eventCallback = null, CancellationToken cancellation = default)
{
Throw.IfNull(handle);
await handle.RunToCompletionAsync(eventCallback, cancellation).ConfigureAwait(false);
return handle.RunningOutput!;
}
}
dotnet/src/Microsoft.Agents.Workflows/Workflow.cs
+42 -46
View File
@@ -3,8 +3,8 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.Agents.Workflows.Checkpointing;
using Microsoft.Agents.Workflows.Specialized;
using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Workflows;
@@ -20,6 +20,7 @@ public class Workflow
internal Dictionary<string, ExecutorRegistration> Registrations { get; init; } = [];
internal Dictionary<string, HashSet<Edge>> Edges { get; init; } = [];
internal HashSet<string> OutputExecutors { get; init; } = [];
/// <summary>
/// Gets the collection of edges grouped by their source node identifier.
@@ -53,21 +54,50 @@ public class Workflow
/// </summary>
public string StartExecutorId { get; }
/// <summary>
/// Gets the type of input expected by the starting executor of the workflow.
/// </summary>
public Type InputType { get; }
/// <summary>
/// Initializes a new instance of the <see cref="Workflow"/> class with the specified starting executor identifier
/// and input type.
/// </summary>
/// <param name="startExecutorId">The unique identifier of the starting executor for the workflow. Cannot be <c>null</c>.</param>
/// <param name="type">The <see cref="Type"/> representing the input data for the workflow. Cannot be <c>null</c>.</param>
internal Workflow(string startExecutorId, Type type)
internal Workflow(string startExecutorId)
{
this.StartExecutorId = Throw.IfNull(startExecutorId);
this.InputType = Throw.IfNull(type);
}
/// <summary>
/// Attempts to promote the current workflow to a type pre-checked instance that can handle input of type <typeparamref name="TInput"/>.
/// </summary>
/// <typeparam name="TInput">The desired input type.</typeparam>
/// <returns>A type-parametrized workflow definitely able to process input of type <typeparamref name="TInput"/> or
/// <see langword="null" /> if the workflow does not accept that type of input.</returns>
/// <exception cref="InvalidOperationException"></exception>
internal async ValueTask<Workflow<TInput>?> TryPromoteAsync<TInput>()
{
// 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.ProviderAsync().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<TInput>(this.StartExecutorId)
{
Registrations = this.Registrations,
Edges = this.Edges,
Ports = this.Ports,
OutputExecutors = this.OutputExecutors
};
}
}
@@ -81,46 +111,12 @@ public class Workflow<T> : Workflow
/// Initializes a new instance of the <see cref="Workflow{T}"/> class with the specified starting executor identifier
/// </summary>
/// <param name="startExecutorId">The unique identifier of the starting executor for the workflow. Cannot be <c>null</c>.</param>
public Workflow(string startExecutorId) : base(startExecutorId, typeof(T))
public Workflow(string startExecutorId) : base(startExecutorId)
{
}
internal Workflow<T, TResult> Promote<TResult>(IOutputSink<TResult> outputSource)
{
Throw.IfNull(outputSource);
return new Workflow<T, TResult>(this.StartExecutorId, outputSource)
{
Registrations = this.Registrations,
Edges = this.Edges,
Ports = this.Ports
};
}
}
/// <summary>
/// Represents a workflow that operates on data of type <typeparamref name="TInput"/>, resulting in
/// <typeparamref name="TResult"/>.
/// </summary>
/// <typeparam name="TInput">The type of input to the workflow.</typeparam>
/// <typeparam name="TResult">The type of the output from the workflow.</typeparam>
public class Workflow<TInput, TResult> : Workflow<TInput>
{
private readonly IOutputSink<TResult> _output;
internal Workflow(string startExecutorId, IOutputSink<TResult> outputSource)
: base(startExecutorId)
{
this._output = Throw.IfNull(outputSource);
}
/// <summary>
/// Gets the unique identifier of the output collector.
/// Gets the type of input expected by the starting executor of the workflow.
/// </summary>
public string OutputCollectorId => this._output.Id;
/// <summary>
/// The running (partial) output of the workflow, if any.
/// </summary>
public TResult? RunningOutput => this._output.Result;
public Type InputType => typeof(T);
}
dotnet/src/Microsoft.Agents.Workflows/WorkflowBuilder.cs
+52 -52
View File
@@ -2,7 +2,6 @@
using System;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
@@ -31,6 +30,7 @@ public class WorkflowBuilder
private readonly HashSet<string> _unboundExecutors = [];
private readonly HashSet<EdgeConnection> _conditionlessConnections = [];
private readonly Dictionary<string, InputPort> _inputPorts = [];
private readonly HashSet<string> _outputExecutors = [];
private readonly string _startExecutorId;
@@ -91,6 +91,23 @@ public class WorkflowBuilder
return executorish;
}
/// <summary>
/// Register executors as an output source. Executors can use <see cref="IWorkflowContext.YieldOutputAsync"/> to yield output values.
/// By default, message handlers with a non-void return type will also be yielded, unless <see cref="ExecutorOptions.AutoYieldOutputHandlerResultObject"/>
/// is set to <see langword="false"/>.
/// </summary>
/// <param name="executors"></param>
/// <returns></returns>
public WorkflowBuilder WithOutputFrom(params ExecutorIsh[] executors)
{
foreach (ExecutorIsh executor in executors)
{
this._outputExecutors.Add(this.Track(executor).Id);
}
return this;
}
/// <summary>
/// Binds the specified executor to the workflow, allowing it to participate in workflow execution.
/// </summary>
@@ -303,41 +320,7 @@ public class WorkflowBuilder
return this;
}
[SuppressMessage("Reliability", "CA2008:Do not create tasks without passing a TaskScheduler",
Justification = "We explicitly set the TaskScheduler when we create the TaskFactory")]
[SuppressMessage("Usage", "VSTHRD002:Avoid problematic synchronous waits",
Justification = "This runs the thread on the thread pool")]
private static TResult RunSync<TResult>(Func<ValueTask<TResult>> funcAsync)
{
TaskFactory factory = new(CancellationToken.None, TaskCreationOptions.None, TaskContinuationOptions.None, TaskScheduler.Default);
// See ASP.Net.Identity's implementation of AsyncHelper
// https://github.com/aspnet/AspNetIdentity/blob/main/src/Microsoft.AspNet.Identity.Core/AsyncHelper.cs
// Capture the current culture and UI culture
var culture = System.Globalization.CultureInfo.CurrentCulture;
var uiCulture = System.Globalization.CultureInfo.CurrentUICulture;
return factory.StartNew(PropagateCultureAndInvokeAsync).Unwrap().GetAwaiter().GetResult();
Task<TResult> PropagateCultureAndInvokeAsync()
{
// Set the culture and UI culture to the captured values
System.Globalization.CultureInfo.CurrentCulture = culture;
System.Globalization.CultureInfo.CurrentUICulture = uiCulture;
return funcAsync().AsTask();
}
}
/// <summary>
/// Builds and returns a workflow instance configured to process messages of the specified input type.
/// </summary>
/// <typeparam name="T">The type of input messages that the workflow will accept and process.</typeparam>
/// <returns>A new instance of <see cref="Workflow{T}"/>.</returns>
/// <exception cref="InvalidOperationException">Thrown if there are unbound executors in the workflow definition,
/// if the start executor is not bound, or if the start executor does not contain a handler for the specified input
/// type <typeparamref name="T"/>.</exception>
public Workflow<T> Build<T>()
private void Validate()
{
if (this._unboundExecutors.Count > 0)
{
@@ -345,27 +328,44 @@ public class WorkflowBuilder
$"Workflow cannot be built because there are unbound executors: {string.Join(", ", this._unboundExecutors)}.");
}
// Grab the start node, and make sure it has the right type?
if (!this._executors.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: This is likely a pipe-dream, but can we do any type-checking on the edges? (Not without instantiating the executors...)
}
Executor startExecutor = RunSync(startRegistration.CreateInstanceAsync);
if (!startExecutor.InputTypes.Any(t => t.IsAssignableFrom(typeof(T))))
{
// 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
throw new InvalidOperationException(
$"Workflow cannot be built because the starting executor {this._startExecutorId} does not contain a handler for the desired input type {typeof(T).Name}");
}
/// <summary>
/// Builds and returns a workflow instance.
/// </summary>
/// <exception cref="InvalidOperationException">Thrown if there are unbound executors in the workflow definition,
/// or if the start executor is not bound.</exception>
public Workflow Build()
{
this.Validate();
return new Workflow<T>(this._startExecutorId) // Why does it not see the default ctor?
return new Workflow(this._startExecutorId)
{
Registrations = this._executors,
Edges = this._edges,
Ports = this._inputPorts
Ports = this._inputPorts,
OutputExecutors = this._outputExecutors
};
}
/// <summary>
/// Attempts to build a workflow instance configured to process messages of the specified input type.
/// </summary>
/// <typeparam name="TInput">The desired input type for the workflow.</typeparam>
/// <exception cref="InvalidOperationException">Thrown if the built workflow cannot process messages of the specified input type,</exception>
public async ValueTask<Workflow<TInput>> BuildAsync<TInput>() where TInput : notnull
{
Workflow<TInput>? maybeWorkflow = await this.Build()
.TryPromoteAsync<TInput>()
.ConfigureAwait(false);
if (maybeWorkflow is null)
{
throw new InvalidOperationException(
$"The built workflow cannot process input of type '{typeof(TInput).FullName}'.");
}
return maybeWorkflow;
}
}
dotnet/src/Microsoft.Agents.Workflows/WorkflowBuilderExtensions.cs
-37
View File
@@ -3,7 +3,6 @@
using System;
using System.Collections.Generic;
using System.Linq;
using Microsoft.Agents.Workflows.Specialized;
using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Workflows;
@@ -157,40 +156,4 @@ public static class WorkflowBuilderExtensions
return switchBuilder.ReduceToFanOut(builder, source);
}
/// <summary>
/// Builds a workflow that collects output from the specified executor, aggregates results using the provided
/// streaming aggregator, and optionally completes based on a custom condition.
/// </summary>
/// <remarks>The returned workflow promotes the output collector as its result source, allowing consumers
/// to access the aggregated output directly. The completion condition can be used to implement custom termination
/// logic, such as early stopping when a desired result is reached.</remarks>
/// <typeparam name="TInput">The type of input items processed by the workflow.</typeparam>
/// <typeparam name="TIntermediate">The type of items generated by the <paramref name="outputSource"/>,
/// and aggregated by the <paramref name="aggregator"/>.</typeparam>
/// <typeparam name="TResult">The type of aggregated result produced by the workflow.</typeparam>
/// <param name="builder">The workflow builder used to construct the workflow and define its execution graph.</param>
/// <param name="outputSource">The executor that produces output items to be collected and aggregated. Cannot be null.</param>
/// <param name="aggregator">The streaming aggregator that processes input items and produces aggregated results. Cannot be null.</param>
/// <param name="completionCondition">An optional predicate that determines when the workflow should complete based on the current input and
/// aggregated result. If null, the workflow will not raise a <see cref="WorkflowCompletedEvent"/>.</param>
/// <returns>A workflow that collects output from the specified executor, aggregates results, and exposes the aggregated
/// output.</returns>
public static Workflow<TInput, TResult> BuildWithOutput<TInput, TIntermediate, TResult>(
this WorkflowBuilder builder,
ExecutorIsh outputSource,
StreamingAggregator<TIntermediate, TResult> aggregator,
Func<TIntermediate, TResult?, bool>? completionCondition = null)
{
Throw.IfNull(outputSource);
Throw.IfNull(aggregator);
OutputCollectorExecutor<TIntermediate, TResult> outputSink = new(aggregator, completionCondition);
// TODO: Check that the outputSource has a TResult output?
builder.AddEdge(outputSource, outputSink);
Workflow<TInput> workflow = builder.Build<TInput>();
return workflow.Promote(outputSink);
}
}
dotnet/src/Microsoft.Agents.Workflows/WorkflowCompletedEvent.cs
-13
View File
@@ -1,13 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
namespace Microsoft.Agents.Workflows;
/// <summary>
/// Event triggered when a workflow completes execution.
/// </summary>
/// <remarks>
/// The user is expected to raise this event from a terminating <see cref="Executor"/>, or to build
/// the workflow with output capture using <see cref="WorkflowBuilderExtensions.BuildWithOutput"/>.
/// </remarks>
/// <param name="result">The result of the execution of the workflow.</param>
public sealed class WorkflowCompletedEvent(object? result = null) : WorkflowEvent(data: result);
dotnet/src/Microsoft.Agents.Workflows/WorkflowEvent.cs
+1 -1
View File
@@ -10,9 +10,9 @@ namespace Microsoft.Agents.Workflows;
[JsonDerivedType(typeof(ExecutorEvent))]
[JsonDerivedType(typeof(SuperStepEvent))]
[JsonDerivedType(typeof(WorkflowStartedEvent))]
[JsonDerivedType(typeof(WorkflowCompletedEvent))]
[JsonDerivedType(typeof(WorkflowErrorEvent))]
[JsonDerivedType(typeof(WorkflowWarningEvent))]
[JsonDerivedType(typeof(WorkflowOutputEvent))]
[JsonDerivedType(typeof(RequestInfoEvent))]
public class WorkflowEvent(object? data = null)
{
dotnet/src/Microsoft.Agents.Workflows/WorkflowHostAgent.cs
+1 -1
View File
@@ -65,7 +65,7 @@ internal sealed class WorkflowHostAgent : AIAgent
// in the case of new threads.
if (!this._runningWorkflows.TryGetValue(runId, out StreamingRun? run))
{
run = await InProcessExecution.StreamAsync(this._workflow, messages, cancellation)
run = await InProcessExecution.StreamAsync(this._workflow, messages, cancellation: cancellation)
.ConfigureAwait(false);
this._runningWorkflows[runId] = run;
}
dotnet/src/Microsoft.Agents.Workflows/WorkflowHostingExtensions.cs
+22
View File
@@ -1,6 +1,8 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents;
@@ -23,6 +25,26 @@ public static class WorkflowHostingExtensions
return new WorkflowHostAgent(workflow, id, name);
}
/// <summary>
/// Convert a workflow with the appropriate primary input type to an <see cref="AIAgent"/>.
/// </summary>
/// <param name="workflow"></param>
/// <param name="id"></param>
/// <param name="name"></param>
/// <returns></returns>
public static async ValueTask<AIAgent> AsAgentAsync(this Workflow workflow, string? id = null, string? name = null)
{
Workflow<List<ChatMessage>>? maybeTyped = await workflow.TryPromoteAsync<List<ChatMessage>>()
.ConfigureAwait(false);
if (maybeTyped is null)
{
throw new InvalidOperationException("Cannot host a workflow that does not accept List<ChatMessage> as an input");
}
return maybeTyped.AsAgent();
}
internal static FunctionCallContent ToFunctionCall(this ExternalRequest request)
{
Dictionary<string, object?> parameters = new()
dotnet/src/Microsoft.Agents.Workflows/WorkflowOutputEvent.cs
+51
View File
@@ -0,0 +1,51 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
namespace Microsoft.Agents.Workflows;
/// <summary>
/// Event triggered when a workflow executor yields output.
/// </summary>
public sealed class WorkflowOutputEvent : WorkflowEvent
{
internal WorkflowOutputEvent(object data, string sourceId) : base(data)
{
this.SourceId = sourceId;
}
/// <summary>
/// The unique identifier of the executor that yielded this output.
/// </summary>
public string SourceId { get; }
/// <summary>
/// Determines whether the underlying data is of the specified type or a derived type.
/// </summary>
/// <typeparam name="T">The type to compare with the type of the underlying data.</typeparam>
/// <returns>true if the underlying data is assignable to type T; otherwise, false.</returns>
public bool Is<T>() => this.IsType(typeof(T));
/// <summary>
/// Determines whether the underlying data is of the specified type or a derived type.
/// </summary>
/// <param name="type">The type to compare with the type of the underlying data.</param>
/// <returns>true if the underlying data is assignable to type T; otherwise, false.</returns>
public bool IsType(Type type) => this.Data == null
? false
: type.IsAssignableFrom(this.Data.GetType());
/// <summary>
/// Attempts to retrieve the underlying data as the specified type.
/// </summary>
/// <typeparam name="T">The type to which to cast.</typeparam>
/// <returns>The value of Data as to the target type.</returns>
public T? As<T>() => this.Data is T value ? value : default;
/// <summary>
/// Attempts to retrieve the underlying data as the specified type.
/// </summary>
/// <param name="type">The type to which to cast.</param>
/// <returns>The value of Data as to the target type.</returns>
public object? AsType(Type type) => this.IsType(type) ? this.Data : null;
}