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

Minor cleanups

Browse Source
This commit is contained in:
Shyju Krishnankutty
2026-01-27 21:00:45 -08:00
Unverified
parent 45050a2b08
commit d0ad92af7a
11 changed files with 117 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dotnet/src/Microsoft.Agents.AI.DurableTask/DurableOptions.cs
+3
View File
@@ -1,10 +1,13 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics;
namespace Microsoft.Agents.AI.DurableTask;
/// <summary>
/// Provides configuration options for durable agents and workflows.
/// </summary>
[DebuggerDisplay("Workflows = {Workflows.Workflows.Count}, Agents = {Agents.AgentCount}")]
public sealed class DurableOptions
{
/// <summary>
dotnet/src/Microsoft.Agents.AI.DurableTask/DurableRequestInfoEvent.cs
+2
View File
@@ -1,5 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Text.Json;
using Microsoft.Agents.AI.Workflows;
@@ -14,6 +15,7 @@ namespace Microsoft.Agents.AI.DurableTask;
/// <param name="RequestType">The full type name of the request type.</param>
/// <param name="ResponseType">The full type name of the expected response type.</param>
/// <param name="RequestPort">The request port definition, if available.</param>
[DebuggerDisplay("RequestPort = {RequestPortId}")]
public sealed class DurableRequestInfoEvent(
string RequestPortId,
string Input,
dotnet/src/Microsoft.Agents.AI.DurableTask/DurableRun.cs
+2
View File
@@ -1,5 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Text.Json;
using Microsoft.Agents.AI.Workflows;
@@ -14,6 +15,7 @@ namespace Microsoft.Agents.AI.DurableTask;
/// This class provides a similar API to <see cref="Run"/> but for workflows executed as durable orchestrations.
/// Events are received by raising external events to the orchestration and can be streamed to the caller.
/// </remarks>
[DebuggerDisplay("{WorkflowName} ({InstanceId})")]
public sealed class DurableRun : IRun
{
private readonly DurableTaskClient _client;
dotnet/src/Microsoft.Agents.AI.DurableTask/DurableStreamingRun.cs
+2
View File
@@ -1,5 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Runtime.CompilerServices;
using System.Text.Json;
@@ -16,6 +17,7 @@ namespace Microsoft.Agents.AI.DurableTask;
/// Events are detected by monitoring the orchestration status for <see cref="RequestPort"/> executors that are waiting
/// for external input (human-in-the-loop scenarios).
/// </remarks>
[DebuggerDisplay("{WorkflowName} ({InstanceId})")]
public sealed class DurableStreamingRun : IStreamingRun
{
private readonly DurableTaskClient _client;
dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowCompletedEvent.cs
+2
View File
@@ -1,5 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics;
using Microsoft.Agents.AI.Workflows;
namespace Microsoft.Agents.AI.DurableTask;
@@ -7,6 +8,7 @@ namespace Microsoft.Agents.AI.DurableTask;
/// <summary>
/// Event raised when a durable workflow completes successfully.
/// </summary>
[DebuggerDisplay("Completed: {Result}")]
public sealed class DurableWorkflowCompletedEvent : WorkflowEvent
{
/// <summary>
dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowFailedEvent.cs
+2
View File
@@ -1,5 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics;
using Microsoft.Agents.AI.Workflows;
namespace Microsoft.Agents.AI.DurableTask;
@@ -7,6 +8,7 @@ namespace Microsoft.Agents.AI.DurableTask;
/// <summary>
/// Event raised when a durable workflow fails.
/// </summary>
[DebuggerDisplay("Failed: {ErrorMessage}")]
public sealed class DurableWorkflowFailedEvent : WorkflowEvent
{
/// <summary>
dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowOptions.cs
+2
View File
@@ -1,5 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics;
using Microsoft.Agents.AI.Workflows;
namespace Microsoft.Agents.AI.DurableTask;
@@ -7,6 +8,7 @@ namespace Microsoft.Agents.AI.DurableTask;
/// <summary>
/// Provides configuration options for managing durable workflows within an application.
/// </summary>
[DebuggerDisplay("Workflows = {Workflows.Count}")]
public sealed class DurableWorkflowOptions
{
private readonly Dictionary<string, Workflow> _workflows = new(StringComparer.OrdinalIgnoreCase);
dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs
+93 -57
View File
@@ -366,28 +366,13 @@ internal class DurableWorkflowRunner
/// <item><description>Maximum superstep limit is reached (safety limit)</description></item>
/// </list>
/// <para>
/// <strong>Message Routing Rules:</strong>
/// <strong>Fan-In Handling:</strong>
/// </para>
/// <para>
/// For executors with multiple predecessors (Fan-In), the implementation waits for all predecessor
/// results before invoking the executor. If the executor accepts an array type, all messages are
/// aggregated into a JSON array.
/// </para>
/// <list type="number">
/// <item>
/// <description>
/// Messages sent via <c>SendMessageAsync</c> take priority and include explicit type information.
/// This is the primary mechanism for void-returning executors.
/// </description>
/// </item>
/// <item>
/// <description>
/// If no messages were sent explicitly, the executor's return value is routed to successors.
/// The type information comes from <see cref="WorkflowExecutionPlan.ExecutorOutputTypes"/>.
/// </description>
/// </item>
/// <item>
/// <description>
/// Edge conditions are evaluated before routing. If a condition returns false, the message
/// is not forwarded to that particular successor.
/// </description>
/// </item>
/// </list>
/// </remarks>
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")]
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")]
@@ -441,54 +426,93 @@ internal class DurableWorkflowRunner
activeExecutors.Count,
string.Join(", ", activeExecutors));
// Process each active executor
// Prepare execution tasks for all active executors (for parallel dispatch)
List<(string ExecutorId, string Input, string? InputTypeName, WorkflowExecutorInfo Info)> executorInputs = [];
foreach (string executorId in activeExecutors)
{
Queue<(string Message, string? InputTypeName)> queue = messageQueues[executorId];
// Process all messages for this executor in this superstep
while (queue.Count > 0)
// Check if this is a Fan-In executor that expects aggregated input
bool isFanIn = plan.Predecessors.TryGetValue(executorId, out List<string>? predecessors) && predecessors.Count > 1;
string input;
string? inputTypeName;
if (isFanIn && queue.Count > 1)
{
(string input, string? inputTypeName) = queue.Dequeue();
// Create executor info
WorkflowExecutorInfo executorInfo = CreateExecutorInfo(executorId, executorBindings);
// Execute the activity with type information
string rawResult = await this.DispatchExecutorAsync(
context, executorInfo, input, inputTypeName, logger, customStatus, sharedState).ConfigureAwait(true);
(string result, List<SentMessageInfo> sentMessages) = UnwrapActivityResult(rawResult, customStatus, sharedState);
lastResults[executorId] = result;
// Check for explicit halt request (via RequestHaltAsync)
if (HasHaltBeenRequested(customStatus, executorId))
// Fan-In: Aggregate all messages into a JSON array
List<string> messages = [];
while (queue.Count > 0)
{
haltRequested = true;
finalOutput = result;
logger.LogDebug("Halt requested by executor {ExecutorId}", executorId);
break;
(string msg, _) = queue.Dequeue();
messages.Add(msg);
}
// Route messages sent via SendMessageAsync (takes priority for void-returning executors)
if (sentMessages.Count > 0)
input = AggregateMessagesToJsonArray(messages);
inputTypeName = typeof(string[]).FullName;
logger.LogDebug("Fan-In executor {ExecutorId}: aggregated {Count} messages", executorId, messages.Count);
}
else
{
// Normal case: process single message
(input, inputTypeName) = queue.Dequeue();
}
WorkflowExecutorInfo executorInfo = CreateExecutorInfo(executorId, executorBindings);
executorInputs.Add((executorId, input, inputTypeName, executorInfo));
}
// Dispatch all executors in parallel by starting all tasks first, then awaiting together
List<Task<string>> executorTasks = [];
foreach ((string executorId, string input, string? inputTypeName, WorkflowExecutorInfo executorInfo) in executorInputs)
{
// Start the task without awaiting - this enables parallel dispatch
Task<string> task = this.DispatchExecutorAsync(
context, executorInfo, input, inputTypeName, logger, customStatus, sharedState);
executorTasks.Add(task);
}
// Wait for all executors to complete in parallel
string[] rawResults = await Task.WhenAll(executorTasks).ConfigureAwait(true);
// Process results and route messages
for (int i = 0; i < executorInputs.Count; i++)
{
string executorId = executorInputs[i].ExecutorId;
string rawResult = rawResults[i];
(string result, List<SentMessageInfo> sentMessages) = UnwrapActivityResult(rawResult, customStatus, sharedState);
lastResults[executorId] = result;
// Check for explicit halt request (via RequestHaltAsync)
if (HasHaltBeenRequested(customStatus, executorId))
{
haltRequested = true;
finalOutput = result;
logger.LogDebug("Halt requested by executor {ExecutorId}", executorId);
break;
}
// Route messages sent via SendMessageAsync (takes priority for void-returning executors)
if (sentMessages.Count > 0)
{
foreach (SentMessageInfo sentMessage in sentMessages)
{
foreach (SentMessageInfo sentMessage in sentMessages)
if (!string.IsNullOrEmpty(sentMessage.Message))
{
if (!string.IsNullOrEmpty(sentMessage.Message))
{
// Route to successors with the sent message's type
RouteMessageToSuccessors(
executorId, sentMessage.Message, sentMessage.TypeName, plan, messageQueues, logger);
}
// Route to successors with the sent message's type
RouteMessageToSuccessors(
executorId, sentMessage.Message, sentMessage.TypeName, plan, messageQueues, logger);
}
}
else if (!string.IsNullOrEmpty(result))
{
// Route executor's return value to successor executors via edges (for non-void executors)
RouteMessageToSuccessors(
executorId, result, plan, messageQueues, logger);
}
}
else if (!string.IsNullOrEmpty(result))
{
// Route executor's return value to successor executors via edges (for non-void executors)
RouteMessageToSuccessors(
executorId, result, plan, messageQueues, logger);
}
if (haltRequested)
@@ -509,6 +533,18 @@ internal class DurableWorkflowRunner
return finalOutput ?? DetermineFinalResult(workflow, lastResults, customStatus);
}
/// <summary>
/// Aggregates multiple messages into a JSON array.
/// </summary>
/// <param name="messages">The messages to aggregate.</param>
/// <returns>A JSON array string containing all messages.</returns>
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing string array.")]
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing string array.")]
private static string AggregateMessagesToJsonArray(List<string> messages)
{
return JsonSerializer.Serialize(messages);
}
/// <summary>
/// Enqueues a message to an executor's message queue with type information.
/// </summary>
dotnet/src/Microsoft.Agents.AI.DurableTask/DurableYieldedOutputEvent.cs
+2
View File
@@ -1,5 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics;
using Microsoft.Agents.AI.Workflows;
namespace Microsoft.Agents.AI.DurableTask;
@@ -11,6 +12,7 @@ namespace Microsoft.Agents.AI.DurableTask;
/// This is the durable equivalent of <see cref="WorkflowOutputEvent"/> since that class has an internal
/// constructor not accessible from outside the Workflows assembly.
/// </remarks>
[DebuggerDisplay("Yielded by {ExecutorId}: {Output}")]
public sealed class DurableYieldedOutputEvent : WorkflowEvent
{
/// <summary>
dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowExecutionPlan.cs
+3 -21
View File
@@ -1,10 +1,13 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics;
namespace Microsoft.Agents.AI.DurableTask;
/// <summary>
/// Represents the complete execution plan for a workflow, including parallel execution levels.
/// </summary>
[DebuggerDisplay("Start = {StartExecutorId}, Levels = {Levels.Count}")]
internal sealed class WorkflowExecutionPlan
{
/// <summary>
@@ -33,27 +36,6 @@ internal sealed class WorkflowExecutionPlan
/// </summary>
public Dictionary<string, Type?> ExecutorOutputTypes { get; } = [];
/// <summary>
/// Gets whether this workflow has any parallel execution opportunities.
/// </summary>
public bool HasParallelism => this.Levels.Any(l => l.Executors.Count > 1);
/// <summary>
/// Gets whether this workflow has any Fan-In points.
/// </summary>
public bool HasFanIn => this.Levels.Any(l => l.IsFanIn);
/// <summary>
/// Gets or sets whether this workflow contains cycles requiring iterative message-driven execution.
/// </summary>
public bool HasCycles { get; set; }
/// <summary>
/// Gets the back-edges that create cycles in the workflow graph.
/// These edges are excluded from topological level computation but are followed during message-driven execution.
/// </summary>
public List<(string SourceId, string TargetId)> BackEdges { get; } = [];
/// <summary>
/// Gets or sets the starting executor ID for the workflow.
/// </summary>
dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs
+4 -8
View File
@@ -1,5 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics;
using Microsoft.Agents.AI.Workflows;
namespace Microsoft.Agents.AI.DurableTask;
@@ -10,6 +11,7 @@ namespace Microsoft.Agents.AI.DurableTask;
/// <param name="ExecutorId">The unique identifier of the executor.</param>
/// <param name="IsAgenticExecutor">Indicates whether this executor is an agentic executor.</param>
/// <param name="RequestPort">The request port if this executor is a request port executor; otherwise, null.</param>
[DebuggerDisplay("{ExecutorId}, Agentic = {IsAgenticExecutor}, HITL = {IsRequestPortExecutor}")]
internal sealed record WorkflowExecutorInfo(string ExecutorId, bool IsAgenticExecutor, RequestPort? RequestPort = null)
{
/// <summary>
@@ -25,6 +27,7 @@ internal sealed record WorkflowExecutorInfo(string ExecutorId, bool IsAgenticExe
/// <param name="Level">The level number (0-based, starting from the root executor).</param>
/// <param name="Executors">The executors that can run in parallel at this level.</param>
/// <param name="IsFanIn">Indicates if this level is a Fan-In point (has executors with multiple predecessors).</param>
[DebuggerDisplay("Level {Level}: {Executors.Count} executor(s), FanIn = {IsFanIn}")]
internal sealed record WorkflowExecutionLevel(int Level, List<WorkflowExecutorInfo> Executors, bool IsFanIn);
/// <summary>
@@ -118,13 +121,6 @@ internal static class WorkflowHelper
// Detect back-edges using DFS from start executor
HashSet<(string Source, string Target)> backEdges = DetectBackEdges(workflow.StartExecutorId, successors);
// Mark cycles in plan
plan.HasCycles = backEdges.Count > 0;
foreach ((string source, string target) in backEdges)
{
plan.BackEdges.Add((source, target));
}
// Calculate in-degrees, EXCLUDING back-edges
int[] inDegree = new int[executors.Count];
foreach (string executorId in executors.Keys)
@@ -145,7 +141,7 @@ internal static class WorkflowHelper
plan.EdgeConditions[condition.Key] = condition.Value;
}
// Store the graph structure in the plan (reuse the built lists directly)
// Store the graph structure in the plan
foreach (string executorId in executors.Keys)
{
plan.Predecessors[executorId] = predecessors[executorId];