agent/agent-framework
1
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

Optimizations.

Browse Source
This commit is contained in:
Shyju Krishnankutty
2026-01-24 16:41:20 -08:00
Unverified
parent 6224bce853
commit ffb2468945
2 changed files with 94 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs
+27 -39
View File
@@ -156,19 +156,15 @@ public class DurableWorkflowRunner
}
string json = input;
if (input.StartsWith('"') && input.EndsWith('"'))
// Check for double-serialized strings (e.g., "\"actual json\"")
// A properly double-serialized string starts with \" and ends with \"
if (input.Length > 2 && input[0] == '"' && input[^1] == '"' && input[1] == '\\')
{
try
string? innerJson = JsonSerializer.Deserialize<string>(input);
if (innerJson is not null)
{
string? innerJson = JsonSerializer.Deserialize<string>(input);
if (innerJson is not null)
{
json = innerJson;
}
}
catch (JsonException)
{
// Not double-serialized, use original
json = innerJson;
}
}
@@ -183,7 +179,7 @@ public class DurableWorkflowRunner
ILogger logger)
{
WorkflowExecutionPlan plan = WorkflowHelper.GetExecutionPlan(workflow);
Dictionary<string, string> results = [];
Dictionary<string, string> results = new(plan.Levels.Sum(l => l.Executors.Count));
foreach (WorkflowExecutionLevel level in plan.Levels)
{
@@ -204,14 +200,16 @@ public class DurableWorkflowRunner
}
else
{
List<Task<(string Id, string Result)>> tasks = [];
foreach (WorkflowExecutorInfo executorInfo in eligibleExecutors)
Task<(string Id, string Result)>[] tasks = new Task<(string Id, string Result)>[eligibleExecutors.Count];
for (int i = 0; i < eligibleExecutors.Count; i++)
{
WorkflowExecutorInfo executorInfo = eligibleExecutors[i];
string input = GetExecutorInput(executorInfo.ExecutorId, initialInput, results, plan);
tasks.Add(this.ExecuteExecutorWithIdAsync(context, executorInfo, input, logger));
tasks[i] = this.ExecuteExecutorWithIdAsync(context, executorInfo, input, logger);
}
foreach ((string id, string result) in await Task.WhenAll(tasks).ConfigureAwait(true))
(string Id, string Result)[] completedTasks = await Task.WhenAll(tasks).ConfigureAwait(true);
foreach ((string id, string result) in completedTasks)
{
results[id] = result;
}
@@ -234,15 +232,13 @@ public class DurableWorkflowRunner
WorkflowExecutionPlan plan,
ILogger logger)
{
List<WorkflowExecutorInfo> eligible = [];
List<WorkflowExecutorInfo> eligible = new(executors.Count);
foreach (WorkflowExecutorInfo executorInfo in executors)
{
List<string> predecessors = plan.Predecessors[executorInfo.ExecutorId];
// Root executor (no predecessors) is always eligible
if (predecessors.Count == 0)
if (!plan.Predecessors.TryGetValue(executorInfo.ExecutorId, out List<string>? predecessors) || predecessors.Count == 0)
{
// Root executor (no predecessors) is always eligible
eligible.Add(executorInfo);
continue;
}
@@ -252,16 +248,9 @@ public class DurableWorkflowRunner
foreach (string predecessorId in predecessors)
{
// Get the condition for this edge (predecessor -> current executor)
if (!plan.EdgeConditions.TryGetValue((predecessorId, executorInfo.ExecutorId), out Func<object?, bool>? condition))
if (!plan.EdgeConditions.TryGetValue((predecessorId, executorInfo.ExecutorId), out Func<object?, bool>? condition) || condition is null)
{
// No condition registered for this edge, assume it's eligible
isEligible = true;
break;
}
if (condition is null)
{
// Edge has no condition, always eligible
// No condition registered for this edge or edge has no condition, always eligible
isEligible = true;
break;
}
@@ -272,7 +261,7 @@ public class DurableWorkflowRunner
try
{
// Get the predecessor's output type for proper deserialization
Type? predecessorOutputType = plan.ExecutorOutputTypes.GetValueOrDefault(predecessorId);
plan.ExecutorOutputTypes.TryGetValue(predecessorId, out Type? predecessorOutputType);
// Deserialize the predecessor result to the expected type for condition evaluation
object? resultObject = DeserializeForCondition(predecessorResult, predecessorOutputType);
@@ -382,9 +371,7 @@ public class DurableWorkflowRunner
Dictionary<string, string> results,
WorkflowExecutionPlan plan)
{
List<string> predecessors = plan.Predecessors[executorId];
if (predecessors.Count == 0)
if (!plan.Predecessors.TryGetValue(executorId, out List<string>? predecessors) || predecessors.Count == 0)
{
return initialInput;
}
@@ -394,7 +381,7 @@ public class DurableWorkflowRunner
return results.TryGetValue(predecessors[0], out string? result) ? result : initialInput;
}
List<string> aggregated = [];
List<string> aggregated = new(predecessors.Count);
foreach (string predecessorId in predecessors)
{
if (results.TryGetValue(predecessorId, out string? result))
@@ -409,14 +396,15 @@ public class DurableWorkflowRunner
private static string GetFinalResult(WorkflowExecutionPlan plan, Dictionary<string, string> results)
{
WorkflowExecutionLevel lastLevel = plan.Levels[^1];
List<WorkflowExecutorInfo> lastExecutors = lastLevel.Executors;
if (lastLevel.Executors.Count == 1)
if (lastExecutors.Count == 1)
{
return results[lastLevel.Executors[0].ExecutorId];
return results.TryGetValue(lastExecutors[0].ExecutorId, out string? singleResult) ? singleResult : string.Empty;
}
List<string> finalResults = [];
foreach (WorkflowExecutorInfo executor in lastLevel.Executors)
List<string> finalResults = new(lastExecutors.Count);
foreach (WorkflowExecutorInfo executor in lastExecutors)
{
if (results.TryGetValue(executor.ExecutorId, out string? result))
{
dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs
+67 -56
View File
@@ -105,16 +105,18 @@ public static class WorkflowHelper
WorkflowExecutionPlan plan = new();
// Build adjacency lists (successors and predecessors)
Dictionary<string, List<string>> successors = [];
Dictionary<string, List<string>> predecessors = [];
Dictionary<string, int> inDegree = [];
Dictionary<string, List<string>> successors = new(executors.Count);
Dictionary<string, List<string>> predecessors = new(executors.Count);
int[] inDegree = new int[executors.Count];
Dictionary<string, int> executorIndex = new(executors.Count);
// Initialize all executors and extract their output types
int index = 0;
foreach (KeyValuePair<string, ExecutorBinding> executor in executors)
{
executorIndex[executor.Key] = index++;
successors[executor.Key] = [];
predecessors[executor.Key] = [];
inDegree[executor.Key] = 0;
// Extract output type from executor type (e.g., Executor<TInput, TOutput> -> TOutput)
plan.ExecutorOutputTypes[executor.Key] = GetExecutorOutputType(executor.Value.ExecutorType);
@@ -124,16 +126,17 @@ public static class WorkflowHelper
foreach (KeyValuePair<string, HashSet<EdgeInfo>> edgeGroup in edges)
{
string sourceId = edgeGroup.Key;
List<string> sourceSuccessors = successors[sourceId];
foreach (EdgeInfo edge in edgeGroup.Value)
{
foreach (string sinkId in edge.Connection.SinkIds)
{
if (executors.ContainsKey(sinkId))
if (executorIndex.TryGetValue(sinkId, out int sinkIdx))
{
successors[sourceId].Add(sinkId);
sourceSuccessors.Add(sinkId);
predecessors[sinkId].Add(sourceId);
inDegree[sinkId]++;
inDegree[sinkIdx]++;
}
}
}
@@ -145,74 +148,82 @@ public static class WorkflowHelper
plan.EdgeConditions[condition.Key] = condition.Value;
}
// Store the graph structure in the plan
// Store the graph structure in the plan (reuse the built lists directly)
foreach (string executorId in executors.Keys)
{
plan.Predecessors[executorId] = [.. predecessors[executorId]];
plan.Successors[executorId] = [.. successors[executorId]];
plan.Predecessors[executorId] = predecessors[executorId];
plan.Successors[executorId] = successors[executorId];
}
// Build execution levels using modified Kahn's algorithm
// Instead of processing one at a time, we process all nodes with in-degree 0 at once (same level)
HashSet<string> processed = [];
Dictionary<string, int> currentInDegree = new(inDegree);
int levelNumber = 0;
while (processed.Count < executors.Count)
// Build execution levels using queue-based Kahn's algorithm
// Process all nodes with in-degree 0 at once (same level) for parallel execution
Queue<string> currentLevel = new();
foreach (KeyValuePair<string, int> kvp in executorIndex)
{
// Find all executors that can be executed at this level (in-degree == 0 and not yet processed)
List<string> currentLevelIds = [];
foreach (KeyValuePair<string, int> kvp in currentInDegree)
if (inDegree[kvp.Value] == 0)
{
if (kvp.Value == 0 && !processed.Contains(kvp.Key))
{
currentLevelIds.Add(kvp.Key);
}
currentLevel.Enqueue(kvp.Key);
}
}
// If no executors found but not all processed, there might be a cycle
if (currentLevelIds.Count == 0)
int levelNumber = 0;
int processedCount = 0;
while (currentLevel.Count > 0)
{
List<WorkflowExecutorInfo> levelExecutors = new(currentLevel.Count);
Queue<string> nextLevel = new();
bool isFanIn = false;
while (currentLevel.Count > 0)
{
// Add remaining unprocessed executors
foreach (string executorId in executors.Keys)
string executorId = currentLevel.Dequeue();
processedCount++;
ExecutorBinding executorBinding = executors[executorId];
bool isAgentic = IsAgentExecutorType(executorBinding.ExecutorType);
levelExecutors.Add(new WorkflowExecutorInfo(executorId, isAgentic));
// Check Fan-In for this executor
if (predecessors[executorId].Count > 1)
{
if (!processed.Contains(executorId))
{
currentLevelIds.Add(executorId);
}
isFanIn = true;
}
if (currentLevelIds.Count == 0)
{
break;
}
}
// Check if this level is a Fan-In point (any executor has multiple predecessors)
bool isFanIn = currentLevelIds.Any(id => predecessors[id].Count > 1);
// Convert to WorkflowExecutorInfo
List<WorkflowExecutorInfo> levelExecutors = [];
foreach (string executorId in currentLevelIds)
{
processed.Add(executorId);
if (executors.TryGetValue(executorId, out ExecutorBinding? executorBinding))
{
bool isAgentic = IsAgentExecutorType(executorBinding.ExecutorType);
levelExecutors.Add(new WorkflowExecutorInfo(executorId, isAgentic));
}
// Decrement in-degree of all successors
// Decrement in-degree of all successors and enqueue those ready for next level
foreach (string successor in successors[executorId])
{
currentInDegree[successor]--;
int successorIdx = executorIndex[successor];
if (--inDegree[successorIdx] == 0)
{
nextLevel.Enqueue(successor);
}
}
}
plan.Levels.Add(new WorkflowExecutionLevel(levelNumber, levelExecutors, isFanIn));
levelNumber++;
currentLevel = nextLevel;
}
// Handle cycle detection: if not all executors were processed, there's a cycle
if (processedCount < executors.Count)
{
List<WorkflowExecutorInfo> remainingExecutors = [];
foreach (KeyValuePair<string, ExecutorBinding> executor in executors)
{
if (inDegree[executorIndex[executor.Key]] > 0)
{
bool isAgentic = IsAgentExecutorType(executor.Value.ExecutorType);
remainingExecutors.Add(new WorkflowExecutorInfo(executor.Key, isAgentic));
}
}
if (remainingExecutors.Count > 0)
{
bool isFanIn = remainingExecutors.Exists(e => predecessors[e.ExecutorId].Count > 1);
plan.Levels.Add(new WorkflowExecutionLevel(levelNumber, remainingExecutors, isFanIn));
}
}
return plan;