diff --git a/dotnet/Directory.Packages.props b/dotnet/Directory.Packages.props index d721e208ff..6ba83def44 100644 --- a/dotnet/Directory.Packages.props +++ b/dotnet/Directory.Packages.props @@ -112,14 +112,14 @@ - - - - + + + + - + diff --git a/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/10_WorkflowConcurrent.csproj b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/10_WorkflowConcurrent.csproj new file mode 100644 index 0000000000..ec1dca7683 --- /dev/null +++ b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/10_WorkflowConcurrent.csproj @@ -0,0 +1,44 @@ + + + net10.0 + v4 + Exe + enable + enable + + SingleAgent + SingleAgent + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/ConcurrentStartExecutor.cs b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/ConcurrentStartExecutor.cs new file mode 100644 index 0000000000..2454ee82dd --- /dev/null +++ b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/ConcurrentStartExecutor.cs @@ -0,0 +1,38 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace SingleAgent; + +internal sealed class ConcurrentStartExecutor() : Executor("ConcurrentStartExecutor") +{ + public override ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // do some initial parsing and validation of the message. + // Return a polished version ith additional metadta. + if (!message.StartsWith("Query for the agent:", StringComparison.OrdinalIgnoreCase)) + { + message = "Query for the agent: " + message; + } + + return ValueTask.FromResult(message); + } +} + +internal sealed class ConcurrentAggregationExecutor() : Executor("ConcurrentAggregationExecutor") +{ + /// + /// Handles incoming messages from the agents and aggregates their responses. + /// + /// The messages from the parallel agents. + /// Workflow context for accessing workflow services and adding events. + /// The to monitor for cancellation requests. + /// The default is . + /// A task representing the asynchronous operation. + public override ValueTask HandleAsync(string[] message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // Aggregate all responses from parallel executors + string aggregatedResponse = string.Join("\n---\n", message); + return ValueTask.FromResult($"Aggregated {message.Length} responses:\n{aggregatedResponse}"); + } +} diff --git a/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/Program.cs b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/Program.cs new file mode 100644 index 0000000000..304ffd9dea --- /dev/null +++ b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/Program.cs @@ -0,0 +1,51 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Azure; +using Azure.AI.OpenAI; +using Azure.Identity; +using Microsoft.Agents.AI; +using Microsoft.Agents.AI.Hosting.AzureFunctions; +using Microsoft.Agents.AI.Workflows; +using Microsoft.Azure.Functions.Worker.Builder; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.Hosting; +using OpenAI.Chat; +using SingleAgent; + +// Get the Azure OpenAI endpoint and deployment name from environment variables. +string endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") + ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set."); +string deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT") + ?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT is not set."); + +// Use Azure Key Credential if provided, otherwise use Azure CLI Credential. +string? azureOpenAiKey = System.Environment.GetEnvironmentVariable("AZURE_OPENAI_KEY"); +AzureOpenAIClient client = !string.IsNullOrEmpty(azureOpenAiKey) + ? new AzureOpenAIClient(new Uri(endpoint), new AzureKeyCredential(azureOpenAiKey)) + : new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential()); + +AIAgent physicist = client.GetChatClient(deploymentName).CreateAIAgent("You are an expert in physics. You answer questions from a physics perspective.", "Physicist"); +AIAgent chemist = client.GetChatClient(deploymentName).CreateAIAgent("You are an expert in chemistry. You answer questions from a chemistry perspective.", "Chemist"); + +var startExecutor = new ConcurrentStartExecutor(); +var aggregationExecutor = new ConcurrentAggregationExecutor(); + +// Build the workflow by adding executors and connecting them +var workflow = new WorkflowBuilder(startExecutor) + .WithName("FanOutWorkflow") + .AddFanOutEdge(startExecutor, [physicist, chemist]) + .AddFanInEdge([physicist, chemist], aggregationExecutor) + .WithOutputFrom(aggregationExecutor) + .Build(); + +// Configure the function app to host AI agents and workflows in a unified way. +// This will automatically generate HTTP API endpoints for agents and workflows. +var functionBuilder = FunctionsApplication.CreateBuilder(args); +functionBuilder + .ConfigureFunctionsWebApplication() + .ConfigureDurableOptions(options => +{ + // Configure workflows + options.Workflows.AddWorkflow(workflow); +}); +functionBuilder.Build().Run(); diff --git a/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/README.md b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/README.md new file mode 100644 index 0000000000..d4ac968978 --- /dev/null +++ b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/README.md @@ -0,0 +1,89 @@ +# Single Agent Sample + +This sample demonstrates how to use the Durable Agent Framework (DAFx) to create a simple Azure Functions app that hosts a single AI agent and provides direct HTTP API access for interactive conversations. + +## Key Concepts Demonstrated + +- Using the Microsoft Agent Framework to define a simple AI agent with a name and instructions. +- Registering agents with the Function app and running them using HTTP. +- Conversation management (via session IDs) for isolated interactions. + +## Environment Setup + +See the [README.md](../README.md) file in the parent directory for more information on how to configure the environment, including how to install and run common sample dependencies. + +## Running the Sample + +With the environment setup and function app running, you can test the sample by sending an HTTP request to the agent endpoint. + +You can use the `demo.http` file to send a message to the agent, or a command line tool like `curl` as shown below: + +Bash (Linux/macOS/WSL): + +```bash +curl -X POST http://localhost:7071/api/agents/Joker/run \ + -H "Content-Type: text/plain" \ + -d "Tell me a joke about a pirate." +``` + +PowerShell: + +```powershell +Invoke-RestMethod -Method Post ` + -Uri http://localhost:7071/api/agents/Joker/run ` + -ContentType text/plain ` + -Body "Tell me a joke about a pirate." +``` + +You can also send JSON requests: + +```bash +curl -X POST http://localhost:7071/api/agents/Joker/run \ + -H "Content-Type: application/json" \ + -H "Accept: application/json" \ + -d '{"message": "Tell me a joke about a pirate."}' +``` + +To continue a conversation, include the `thread_id` in the query string or JSON body: + +```bash +curl -X POST "http://localhost:7071/api/agents/Joker/run?thread_id=your-thread-id" \ + -H "Content-Type: application/json" \ + -H "Accept: application/json" \ + -d '{"message": "Tell me another one."}' +``` + +The response from the agent will be displayed in the terminal where you ran `func start`. The expected `text/plain` output will look something like: + +```text +Why don't pirates ever learn the alphabet? Because they always get stuck at "C"! +``` + +The expected `application/json` output will look something like: + +```json +{ + "status": 200, + "thread_id": "ee6e47a0-f24b-40b1-ade8-16fcebb9eb40", + "response": { + "Messages": [ + { + "AuthorName": "Joker", + "CreatedAt": "2025-11-11T12:00:00.0000000Z", + "Role": "assistant", + "Contents": [ + { + "Type": "text", + "Text": "Why don't pirates ever learn the alphabet? Because they always get stuck at 'C'!" + } + ] + } + ], + "Usage": { + "InputTokenCount": 78, + "OutputTokenCount": 36, + "TotalTokenCount": 114 + } + } +} +``` diff --git a/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/ResponseRouterExecutor.cs b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/ResponseRouterExecutor.cs new file mode 100644 index 0000000000..c4c58c5a60 --- /dev/null +++ b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/ResponseRouterExecutor.cs @@ -0,0 +1,27 @@ +//// Copyright (c) Microsoft. All rights reserved. + +//using Microsoft.Agents.AI.Workflows; + +//namespace SingleAgent; + +///// +///// Routes survey responses to appropriate teams based on rating and category. +///// +//public sealed class ResponseRouterExecutor() : Executor("ResponseRouterExecutor") +//{ +// public override ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) +// { +// if (message.Contains("billing", StringComparison.OrdinalIgnoreCase)) +// { +// return ValueTask.FromResult("Routed to Billing Team"); +// } +// else if (message.Contains("technical", StringComparison.OrdinalIgnoreCase)) +// { +// return ValueTask.FromResult("Routed to Technical Support Team"); +// } +// else +// { +// return ValueTask.FromResult("Routed to General Support Team"); +// } +// } +//} diff --git a/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/SurveyResponseParserExecutor.cs b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/SurveyResponseParserExecutor.cs new file mode 100644 index 0000000000..4b3344a8c1 --- /dev/null +++ b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/SurveyResponseParserExecutor.cs @@ -0,0 +1,82 @@ +//// Copyright (c) Microsoft. All rights reserved. + +//using System.Text.Json; +//using System.Text.RegularExpressions; +//using Microsoft.Agents.AI.Workflows; + +//namespace SingleAgent; + +///// +///// This executor parses survey responses and produces structured output. +///// Example input: "Rating: 8. The app is good but checkout process is confusing." +///// +//[System.Diagnostics.CodeAnalysis.SuppressMessage("Performance", "CA1812:Avoid uninstantiated internal classes", Justification = "Instantiated by workflow framework")] +//internal sealed partial class SurveyResponseParserExecutor() : Executor("SurveyResponseParserExecutor") +//{ +// private static readonly JsonSerializerOptions s_jsonOptions = new() +// { +// WriteIndented = true +// }; + +// [GeneratedRegex(@"Rating:\s*(\d+)", RegexOptions.IgnoreCase)] +// private static partial Regex RatingRegex(); + +// public override ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) +// { +// SurveyResponse response = this.ParseSurveyResponse(message); +// string jsonResult = JsonSerializer.Serialize(response, s_jsonOptions); +// return ValueTask.FromResult(jsonResult); +// } + +// private SurveyResponse ParseSurveyResponse(string message) +// { +// // Parse the message to extract rating and comment +// int? rating = null; +// string comment = message; + +// // Try to extract rating using pattern "Rating: {number}" +// Match ratingMatch = RatingRegex().Match(message); +// if (ratingMatch.Success && int.TryParse(ratingMatch.Groups[1].Value, out int parsedRating)) +// { +// rating = parsedRating; + +// // Remove the rating part from the message to get the comment +// // Find the position after the rating number +// int ratingEndIndex = ratingMatch.Index + ratingMatch.Length; + +// // Skip any separators (period, comma, dash, etc.) and whitespace +// while (ratingEndIndex < message.Length && +// (char.IsWhiteSpace(message[ratingEndIndex]) || +// message[ratingEndIndex] == '.' || +// message[ratingEndIndex] == ',' || +// message[ratingEndIndex] == '-')) +// { +// ratingEndIndex++; +// } + +// if (ratingEndIndex < message.Length) +// { +// comment = message[ratingEndIndex..].Trim(); +// } +// else +// { +// comment = string.Empty; +// } +// } + +// // Create and return the structured response +// return new SurveyResponse +// { +// Rating = rating, +// Comment = comment, +// OriginalMessage = message +// }; +// } + +// private sealed class SurveyResponse +// { +// public int? Rating { get; set; } +// public string Comment { get; set; } = string.Empty; +// public string OriginalMessage { get; set; } = string.Empty; +// } +//} diff --git a/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/demo.http b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/demo.http new file mode 100644 index 0000000000..618a1c5483 --- /dev/null +++ b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/demo.http @@ -0,0 +1,8 @@ +# Default endpoint address for local testing +@authority=http://localhost:7071 + +### Prompt the agent +POST {{authority}}/api/workflows/FanOutWorkflow/run +Content-Type: text/plain + +What is temperature? diff --git a/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/host.json b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/host.json new file mode 100644 index 0000000000..9384a0a583 --- /dev/null +++ b/dotnet/samples/AzureFunctions/10_WorkflowConcurrent/host.json @@ -0,0 +1,20 @@ +{ + "version": "2.0", + "logging": { + "logLevel": { + "Microsoft.Agents.AI.DurableTask": "Information", + "Microsoft.Agents.AI.Hosting.AzureFunctions": "Information", + "DurableTask": "Information", + "Microsoft.DurableTask": "Information" + } + }, + "extensions": { + "durableTask": { + "hubName": "default", + "storageProvider": { + "type": "AzureManaged", + "connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING" + } + } + } +} diff --git a/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/11_WorkflowWithDifferentTypes.csproj b/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/11_WorkflowWithDifferentTypes.csproj new file mode 100644 index 0000000000..ec1dca7683 --- /dev/null +++ b/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/11_WorkflowWithDifferentTypes.csproj @@ -0,0 +1,44 @@ + + + net10.0 + v4 + Exe + enable + enable + + SingleAgent + SingleAgent + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/OrderProcessingExecutors.cs b/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/OrderProcessingExecutors.cs new file mode 100644 index 0000000000..cc0dfef6b8 --- /dev/null +++ b/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/OrderProcessingExecutors.cs @@ -0,0 +1,215 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace SingleAgent; + +/// +/// Represents the details of a customer order. +/// +public sealed class OrderDetails +{ + public int OrderId { get; set; } + + public string CustomerName { get; set; } = string.Empty; + + public string CustomerEmail { get; set; } = string.Empty; + + public List Items { get; set; } = []; + + public decimal TotalAmount { get; set; } + + public OrderStatus Status { get; set; } + + public DateTime OrderDate { get; set; } + + public DateTime? EstimatedDelivery { get; set; } +} + +/// +/// Represents an item in an order. +/// +public sealed class OrderItem +{ + public string ProductName { get; set; } = string.Empty; + + public int Quantity { get; set; } + + public decimal UnitPrice { get; set; } + + public decimal Total => this.Quantity * this.UnitPrice; +} + +/// +/// Represents the status of an order. +/// +public enum OrderStatus +{ + Pending, + Processing, + Shipped, + Delivered, + Cancelled +} + +/// +/// Executor that looks up order details by order ID. +/// Input: int (orderId) +/// Output: OrderDetails +/// +internal sealed class OrderLookupExecutor() : Executor("OrderLookupExecutor") +{ + // Simulated order database + private static readonly Dictionary s_orders = new() + { + [1001] = new OrderDetails + { + OrderId = 1001, + CustomerName = "Alice Johnson", + CustomerEmail = "alice@example.com", + Items = + [ + new OrderItem { ProductName = "Wireless Headphones", Quantity = 1, UnitPrice = 79.99m }, + new OrderItem { ProductName = "Phone Case", Quantity = 2, UnitPrice = 15.99m } + ], + TotalAmount = 111.97m, + Status = OrderStatus.Shipped, + OrderDate = DateTime.UtcNow.AddDays(-3), + EstimatedDelivery = DateTime.UtcNow.AddDays(2) + }, + [1002] = new OrderDetails + { + OrderId = 1002, + CustomerName = "Bob Smith", + CustomerEmail = "bob@example.com", + Items = + [ + new OrderItem { ProductName = "Laptop Stand", Quantity = 1, UnitPrice = 49.99m } + ], + TotalAmount = 49.99m, + Status = OrderStatus.Processing, + OrderDate = DateTime.UtcNow.AddDays(-1), + EstimatedDelivery = DateTime.UtcNow.AddDays(5) + }, + [1003] = new OrderDetails + { + OrderId = 1003, + CustomerName = "Carol Davis", + CustomerEmail = "carol@example.com", + Items = + [ + new OrderItem { ProductName = "USB-C Hub", Quantity = 1, UnitPrice = 35.00m }, + new OrderItem { ProductName = "HDMI Cable", Quantity = 3, UnitPrice = 12.99m }, + new OrderItem { ProductName = "Webcam", Quantity = 1, UnitPrice = 89.99m } + ], + TotalAmount = 163.96m, + Status = OrderStatus.Delivered, + OrderDate = DateTime.UtcNow.AddDays(-7), + EstimatedDelivery = null + } + }; + + public override ValueTask HandleAsync(int message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + if (s_orders.TryGetValue(message, out OrderDetails? order)) + { + return ValueTask.FromResult(order); + } + + // Return a "not found" order + return ValueTask.FromResult(new OrderDetails + { + OrderId = message, + CustomerName = "Unknown", + Status = OrderStatus.Cancelled, + OrderDate = DateTime.UtcNow + }); + } +} + +/// +/// Executor that generates a human-readable summary from order details. +/// Input: OrderDetails +/// Output: string +/// +internal sealed class OrderSummaryExecutor() : Executor("OrderSummaryExecutor") +{ + public override ValueTask HandleAsync(OrderDetails message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + if (message.CustomerName == "Unknown") + { + return ValueTask.FromResult($"❌ Order #{message.OrderId} was not found in our system."); + } + + string statusEmoji = message.Status switch + { + OrderStatus.Pending => "⏳", + OrderStatus.Processing => "🔄", + OrderStatus.Shipped => "📦", + OrderStatus.Delivered => "✅", + OrderStatus.Cancelled => "❌", + _ => "❓" + }; + + string itemsList = string.Join("\n", message.Items.Select(i => + $" • {i.ProductName} (x{i.Quantity}) - ${i.Total:F2}")); + + string deliveryInfo = message.Status == OrderStatus.Delivered + ? "Delivered!" + : message.EstimatedDelivery.HasValue + ? $"Expected: {message.EstimatedDelivery.Value:MMM dd, yyyy}" + : "Calculating..."; + + string summary = $""" + ═══════════════════════════════════════ + 📋 ORDER SUMMARY - #{message.OrderId} + ═══════════════════════════════════════ + + 👤 Customer: {message.CustomerName} + 📧 Email: {message.CustomerEmail} + 📅 Order Date: {message.OrderDate:MMM dd, yyyy} + + {statusEmoji} Status: {message.Status} + 🚚 Delivery: {deliveryInfo} + + ───────────────────────────────────────── + 📦 ITEMS: + {itemsList} + ───────────────────────────────────────── + + 💰 TOTAL: ${message.TotalAmount:F2} + ═══════════════════════════════════════ + """; + + return ValueTask.FromResult(summary); + } +} + +/// +/// Executor that parses a string input to extract an order ID. +/// Input: string (e.g., "Check order 1001" or just "1001") +/// Output: int (orderId) +/// +internal sealed class OrderIdParserExecutor() : Executor("OrderIdParserExecutor") +{ + public override ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // Try to extract order ID from the message + // Handles formats like: "1001", "order 1001", "Check order #1001", etc. + string cleanedInput = message + .Replace("order", "", StringComparison.OrdinalIgnoreCase) + .Replace("#", "") + .Trim(); + + // Find the first number in the string + string numberStr = new(cleanedInput.Where(char.IsDigit).ToArray()); + + if (int.TryParse(numberStr, out int orderId)) + { + return ValueTask.FromResult(orderId); + } + + // Default to an invalid order ID if parsing fails + return ValueTask.FromResult(-1); + } +} diff --git a/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/Program.cs b/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/Program.cs new file mode 100644 index 0000000000..183dd6d2dd --- /dev/null +++ b/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/Program.cs @@ -0,0 +1,38 @@ +// Copyright (c) Microsoft. All rights reserved. + +// This sample demonstrates a workflow with different input/output types: +// - OrderIdParserExecutor: string → int +// - OrderLookupExecutor: int → OrderDetails (custom POCO) +// - OrderSummaryExecutor: OrderDetails → string +// +// Workflow: HTTP Request (string) → Parse Order ID → Lookup Order → Generate Summary + +using Microsoft.Agents.AI.Hosting.AzureFunctions; +using Microsoft.Agents.AI.Workflows; +using Microsoft.Azure.Functions.Worker.Builder; +using Microsoft.Extensions.Hosting; +using SingleAgent; + +// Create the executors with different input/output types +OrderIdParserExecutor orderIdParser = new(); // string → int +OrderLookupExecutor orderLookup = new(); // int → OrderDetails +OrderSummaryExecutor orderSummary = new(); // OrderDetails → string + +// Build the workflow: Parse → Lookup → Summarize +Workflow workflow = new WorkflowBuilder(orderIdParser) + .WithName("OrderLookupWorkflow") + .WithDescription("Looks up an order by ID and returns a formatted summary") + .AddEdge(orderIdParser, orderLookup) // string → int → OrderDetails + .AddEdge(orderLookup, orderSummary) // OrderDetails → string + .WithOutputFrom(orderSummary) + .Build(); + +// Configure the function app to host workflows. +// This will automatically generate HTTP API endpoints for the workflow. +FunctionsApplicationBuilder functionBuilder = FunctionsApplication.CreateBuilder(args); +functionBuilder.ConfigureFunctionsWebApplication().ConfigureDurableOptions(options => +{ + // Register the workflow + options.Workflows.AddWorkflow(workflow); +}); +functionBuilder.Build().Run(); diff --git a/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/README.md b/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/README.md new file mode 100644 index 0000000000..d4ac968978 --- /dev/null +++ b/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/README.md @@ -0,0 +1,89 @@ +# Single Agent Sample + +This sample demonstrates how to use the Durable Agent Framework (DAFx) to create a simple Azure Functions app that hosts a single AI agent and provides direct HTTP API access for interactive conversations. + +## Key Concepts Demonstrated + +- Using the Microsoft Agent Framework to define a simple AI agent with a name and instructions. +- Registering agents with the Function app and running them using HTTP. +- Conversation management (via session IDs) for isolated interactions. + +## Environment Setup + +See the [README.md](../README.md) file in the parent directory for more information on how to configure the environment, including how to install and run common sample dependencies. + +## Running the Sample + +With the environment setup and function app running, you can test the sample by sending an HTTP request to the agent endpoint. + +You can use the `demo.http` file to send a message to the agent, or a command line tool like `curl` as shown below: + +Bash (Linux/macOS/WSL): + +```bash +curl -X POST http://localhost:7071/api/agents/Joker/run \ + -H "Content-Type: text/plain" \ + -d "Tell me a joke about a pirate." +``` + +PowerShell: + +```powershell +Invoke-RestMethod -Method Post ` + -Uri http://localhost:7071/api/agents/Joker/run ` + -ContentType text/plain ` + -Body "Tell me a joke about a pirate." +``` + +You can also send JSON requests: + +```bash +curl -X POST http://localhost:7071/api/agents/Joker/run \ + -H "Content-Type: application/json" \ + -H "Accept: application/json" \ + -d '{"message": "Tell me a joke about a pirate."}' +``` + +To continue a conversation, include the `thread_id` in the query string or JSON body: + +```bash +curl -X POST "http://localhost:7071/api/agents/Joker/run?thread_id=your-thread-id" \ + -H "Content-Type: application/json" \ + -H "Accept: application/json" \ + -d '{"message": "Tell me another one."}' +``` + +The response from the agent will be displayed in the terminal where you ran `func start`. The expected `text/plain` output will look something like: + +```text +Why don't pirates ever learn the alphabet? Because they always get stuck at "C"! +``` + +The expected `application/json` output will look something like: + +```json +{ + "status": 200, + "thread_id": "ee6e47a0-f24b-40b1-ade8-16fcebb9eb40", + "response": { + "Messages": [ + { + "AuthorName": "Joker", + "CreatedAt": "2025-11-11T12:00:00.0000000Z", + "Role": "assistant", + "Contents": [ + { + "Type": "text", + "Text": "Why don't pirates ever learn the alphabet? Because they always get stuck at 'C'!" + } + ] + } + ], + "Usage": { + "InputTokenCount": 78, + "OutputTokenCount": 36, + "TotalTokenCount": 114 + } + } +} +``` diff --git a/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/ResponseRouterExecutor.cs b/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/ResponseRouterExecutor.cs new file mode 100644 index 0000000000..fdf1090ec2 --- /dev/null +++ b/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/ResponseRouterExecutor.cs @@ -0,0 +1,27 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace SingleAgent; + +/// +/// Routes survey responses to appropriate teams based on rating and category. +/// +public sealed class ResponseRouterExecutor() : Executor("ResponseRouterExecutor") +{ + public override ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + if (message.Contains("billing", StringComparison.OrdinalIgnoreCase)) + { + return ValueTask.FromResult("Routed to Billing Team"); + } + else if (message.Contains("technical", StringComparison.OrdinalIgnoreCase)) + { + return ValueTask.FromResult("Routed to Technical Support Team"); + } + else + { + return ValueTask.FromResult("Routed to General Support Team"); + } + } +} diff --git a/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/SurveyResponseParserExecutor.cs b/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/SurveyResponseParserExecutor.cs new file mode 100644 index 0000000000..f6a66c9038 --- /dev/null +++ b/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/SurveyResponseParserExecutor.cs @@ -0,0 +1,82 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Text.Json; +using System.Text.RegularExpressions; +using Microsoft.Agents.AI.Workflows; + +namespace SingleAgent; + +/// +/// This executor parses survey responses and produces structured output. +/// Example input: "Rating: 8. The app is good but checkout process is confusing." +/// +[System.Diagnostics.CodeAnalysis.SuppressMessage("Performance", "CA1812:Avoid uninstantiated internal classes", Justification = "Instantiated by workflow framework")] +internal sealed partial class SurveyResponseParserExecutor() : Executor("SurveyResponseParserExecutor") +{ + private static readonly JsonSerializerOptions s_jsonOptions = new() + { + WriteIndented = true + }; + + [GeneratedRegex(@"Rating:\s*(\d+)", RegexOptions.IgnoreCase)] + private static partial Regex RatingRegex(); + + public override ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + SurveyResponse response = this.ParseSurveyResponse(message); + string jsonResult = JsonSerializer.Serialize(response, s_jsonOptions); + return ValueTask.FromResult(jsonResult); + } + + private SurveyResponse ParseSurveyResponse(string message) + { + // Parse the message to extract rating and comment + int? rating = null; + string comment = message; + + // Try to extract rating using pattern "Rating: {number}" + Match ratingMatch = RatingRegex().Match(message); + if (ratingMatch.Success && int.TryParse(ratingMatch.Groups[1].Value, out int parsedRating)) + { + rating = parsedRating; + + // Remove the rating part from the message to get the comment + // Find the position after the rating number + int ratingEndIndex = ratingMatch.Index + ratingMatch.Length; + + // Skip any separators (period, comma, dash, etc.) and whitespace + while (ratingEndIndex < message.Length && + (char.IsWhiteSpace(message[ratingEndIndex]) || + message[ratingEndIndex] == '.' || + message[ratingEndIndex] == ',' || + message[ratingEndIndex] == '-')) + { + ratingEndIndex++; + } + + if (ratingEndIndex < message.Length) + { + comment = message[ratingEndIndex..].Trim(); + } + else + { + comment = string.Empty; + } + } + + // Create and return the structured response + return new SurveyResponse + { + Rating = rating, + Comment = comment, + OriginalMessage = message + }; + } + + private sealed class SurveyResponse + { + public int? Rating { get; set; } + public string Comment { get; set; } = string.Empty; + public string OriginalMessage { get; set; } = string.Empty; + } +} diff --git a/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/demo.http b/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/demo.http new file mode 100644 index 0000000000..e6275a79da --- /dev/null +++ b/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/demo.http @@ -0,0 +1,26 @@ +# Default endpoint address for local testing +@authority=http://localhost:7071 + +### Look up Order #1001 (Shipped order with multiple items) +POST {{authority}}/api/workflows/OrderLookupWorkflow/run +Content-Type: text/plain + +Check order 1001 + +### Look up Order #1002 (Processing order) +POST {{authority}}/api/workflows/OrderLookupWorkflow/run +Content-Type: text/plain + +order 1002 + +### Look up Order #1003 (Delivered order) +POST {{authority}}/api/workflows/OrderLookupWorkflow/run +Content-Type: text/plain + +1003 + +### Look up non-existent order +POST {{authority}}/api/workflows/OrderLookupWorkflow/run +Content-Type: text/plain + +order #9999 diff --git a/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/host.json b/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/host.json new file mode 100644 index 0000000000..9384a0a583 --- /dev/null +++ b/dotnet/samples/AzureFunctions/11_WorkflowWithDifferentTypes/host.json @@ -0,0 +1,20 @@ +{ + "version": "2.0", + "logging": { + "logLevel": { + "Microsoft.Agents.AI.DurableTask": "Information", + "Microsoft.Agents.AI.Hosting.AzureFunctions": "Information", + "DurableTask": "Information", + "Microsoft.DurableTask": "Information" + } + }, + "extensions": { + "durableTask": { + "hubName": "default", + "storageProvider": { + "type": "AzureManaged", + "connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING" + } + } + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableAgentsOptions.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableAgentsOptions.cs index f08d114121..15a5cae43d 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableAgentsOptions.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableAgentsOptions.cs @@ -10,6 +10,7 @@ public sealed class DurableAgentsOptions // Agent names are case-insensitive private readonly Dictionary> _agentFactories = new(StringComparer.OrdinalIgnoreCase); private readonly Dictionary _agentTimeToLive = new(StringComparer.OrdinalIgnoreCase); + private readonly HashSet _workflowOnlyAgents = new(StringComparer.OrdinalIgnoreCase); /// /// Initializes a new instance of the class. @@ -104,6 +105,22 @@ public sealed class DurableAgentsOptions /// Thrown when is null or whitespace or when an agent with the same name has already been registered. /// public DurableAgentsOptions AddAIAgent(AIAgent agent, TimeSpan? timeToLive = null) + { + return this.AddAIAgent(agent, workflowOnly: false, timeToLive); + } + + /// + /// Adds an AI agent to the options with workflow-only configuration. + /// + /// The agent to add. + /// If true, the agent is only accessible within workflows and won't have HTTP triggers. + /// Optional time-to-live for this agent's entities. If not specified, uses . + /// The options instance. + /// Thrown when is null. + /// + /// Thrown when is null or whitespace or when an agent with the same name has already been registered. + /// + public DurableAgentsOptions AddAIAgent(AIAgent agent, bool workflowOnly, TimeSpan? timeToLive = null) { ArgumentNullException.ThrowIfNull(agent); @@ -123,6 +140,11 @@ public sealed class DurableAgentsOptions this._agentTimeToLive[agent.Name] = timeToLive; } + if (workflowOnly) + { + this._workflowOnlyAgents.Add(agent.Name); + } + return this; } @@ -144,4 +166,14 @@ public sealed class DurableAgentsOptions { return this._agentTimeToLive.TryGetValue(agentName, out TimeSpan? ttl) ? ttl : this.DefaultTimeToLive; } + + /// + /// Determines whether an agent is configured as workflow-only (no HTTP triggers). + /// + /// The name of the agent. + /// True if the agent is workflow-only; otherwise, false. + internal bool IsWorkflowOnly(string agentName) + { + return this._workflowOnlyAgents.Contains(agentName); + } } diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptions.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableOptions.cs similarity index 79% rename from dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptions.cs rename to dotnet/src/Microsoft.Agents.AI.DurableTask/DurableOptions.cs index 71f6b741ff..82bb8a419f 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptions.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableOptions.cs @@ -1,11 +1,9 @@ // Copyright (c) Microsoft. All rights reserved. -using Microsoft.Agents.AI.DurableTask; - -namespace Microsoft.Agents.AI.Hosting.AzureFunctions; +namespace Microsoft.Agents.AI.DurableTask; /// -/// Provides configuration options for durable features in Azure Functions. +/// Provides configuration options for durable agents and workflows. /// public sealed class DurableOptions { diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowOptions.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowOptions.cs index ab8bfcd4b1..8eea0591ce 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowOptions.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowOptions.cs @@ -10,17 +10,31 @@ namespace Microsoft.Agents.AI.DurableTask; public sealed class DurableWorkflowOptions { private readonly Dictionary _workflows = new(StringComparer.OrdinalIgnoreCase); - private readonly object? _parentOptions; + private readonly DurableOptions? _parentOptions; /// /// Initializes a new instance of the class. /// /// Optional parent options container for accessing related configuration. - public DurableWorkflowOptions(object? parentOptions = null) + internal DurableWorkflowOptions(DurableOptions? parentOptions = null) { this._parentOptions = parentOptions; + this.Executors = new ExecutorRegistry(); } + /// + /// Gets the collection of workflows available in the current context, keyed by their unique names. + /// + /// The returned dictionary is read-only and reflects the current set of registered workflows. + /// Changes to the underlying workflow collection are immediately visible through this property. Accessing a + /// workflow by name that does not exist will result in a KeyNotFoundException. + public IReadOnlyDictionary Workflows => this._workflows; + + /// + /// Gets the executor registry. + /// + internal ExecutorRegistry Executors { get; } + /// /// Adds a workflow to the collection for processing or execution. /// @@ -40,6 +54,9 @@ public sealed class DurableWorkflowOptions this._workflows[workflow.Name] = workflow; + // Register executors in the registry for direct lookup + RegisterExecutors(workflow, this.Executors); + // Register any agentic executors with DurableAgentsOptions if available through parent DurableAgentsOptions? agentOptions = this.TryGetAgentOptions(); if (agentOptions is not null) @@ -48,36 +65,9 @@ public sealed class DurableWorkflowOptions } } - /// - /// Gets the collection of workflows available in the current context, keyed by their unique names. - /// - /// The returned dictionary is read-only and reflects the current set of registered workflows. - /// Changes to the underlying workflow collection are immediately visible through this property. Accessing a - /// workflow by name that does not exist will result in a KeyNotFoundException. - public IReadOnlyDictionary Workflows => this._workflows; - - [System.Diagnostics.CodeAnalysis.UnconditionalSuppressMessage("ReflectionAnalysis", "IL2075", - Justification = "Reflection is used to access Agents property from parent options container for automatic agent registration.")] private DurableAgentsOptions? TryGetAgentOptions() { - // Try to extract DurableAgentsOptions from the parent container - // This uses reflection to access the Agents property if available - if (this._parentOptions is null) - { - return null; - } - - // Check if parent has an Agents property (DurableOptions pattern) - System.Reflection.PropertyInfo? agentsProperty = this._parentOptions.GetType() - .GetProperty("Agents", System.Reflection.BindingFlags.Public | System.Reflection.BindingFlags.Instance); - - if (agentsProperty?.PropertyType == typeof(DurableAgentsOptions)) - { - return agentsProperty.GetValue(this._parentOptions) as DurableAgentsOptions; - } - - // If parent is directly DurableAgentsOptions (for backward compatibility) - return this._parentOptions as DurableAgentsOptions; + return this._parentOptions?.Agents; } private static void RegisterAgenticExecutors(Workflow workflow, DurableAgentsOptions agentOptions) @@ -87,8 +77,8 @@ public sealed class DurableWorkflowOptions { try { - // Register the agent with DurableAgentsOptions - agentOptions.AddAIAgent(agent); + // Register the agent as workflow-only (no HTTP trigger) + agentOptions.AddAIAgent(agent, workflowOnly: true); } catch (ArgumentException) { @@ -97,4 +87,15 @@ public sealed class DurableWorkflowOptions } } } + + private static void RegisterExecutors(Workflow workflow, ExecutorRegistry registry) + { + // Register all executors from the workflow in the registry + foreach (KeyValuePair executor in workflow.ReflectExecutors()) + { + // Extract the executor name (without GUID suffix) + string executorName = executor.Key.Split('_')[0]; + registry.Register(executorName, executor.Key, workflow); + } + } } diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/ExecutorRegistry.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/ExecutorRegistry.cs new file mode 100644 index 0000000000..b4d3bd70d4 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/ExecutorRegistry.cs @@ -0,0 +1,75 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace Microsoft.Agents.AI.DurableTask; + +/// +/// Provides a registry for storing and retrieving executor bindings independently from workflows. +/// +/// +/// This registry allows executors to be looked up by name without needing to search through all workflows, +/// which is useful for activity function execution where only the executor name is known. +/// +internal sealed class ExecutorRegistry +{ + private readonly Dictionary _executors = new(StringComparer.OrdinalIgnoreCase); + + /// + /// Registers an executor binding from a workflow. + /// + /// The executor name (without GUID suffix). + /// The full executor ID (may include GUID suffix). + /// The workflow containing the executor. + /// + /// If an executor with the same name is already registered, it will be skipped. + /// This is expected behavior when the same executor is used across multiple workflows. + /// + internal void Register(string executorName, string executorId, Workflow workflow) + { + ArgumentException.ThrowIfNullOrEmpty(executorName); + ArgumentException.ThrowIfNullOrEmpty(executorId); + ArgumentNullException.ThrowIfNull(workflow); + + // Only register if not already present (first registration wins) + if (!this._executors.ContainsKey(executorName)) + { + this._executors[executorName] = new ExecutorRegistration(executorId, workflow); + } + } + + /// + /// Attempts to get an executor registration by name. + /// + /// The executor name to look up. + /// When this method returns, contains the registration if found; otherwise, null. + /// true if the executor was found; otherwise, false. + public bool TryGetExecutor(string executorName, out ExecutorRegistration? registration) + { + return this._executors.TryGetValue(executorName, out registration); + } + + /// + /// Gets the number of registered executors. + /// + public int Count => this._executors.Count; +} + +/// +/// Represents a registered executor with its associated workflow. +/// +/// The full executor ID (may include GUID suffix). +/// The workflow containing the executor. +internal sealed record ExecutorRegistration(string ExecutorId, Workflow Workflow) +{ + /// + /// Creates an instance of the executor. + /// + /// A unique identifier for the run context. + /// The cancellation token. + /// The created executor instance. + public ValueTask CreateExecutorInstanceAsync(string runId, CancellationToken cancellationToken = default) + { + return this.Workflow.CreateExecutorInstanceAsync(this.ExecutorId, runId, cancellationToken); + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableAgentsOptionsExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableAgentsOptionsExtensions.cs index ad21d8f4e1..d3a2fed7a6 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableAgentsOptionsExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableAgentsOptionsExtensions.cs @@ -113,6 +113,29 @@ public static class DurableAgentsOptionsExtensions Func factory, bool enableHttpTrigger, bool enableMcpToolTrigger) + { + return AddAIAgentFactory(options, name, factory, enableHttpTrigger, enableMcpToolTrigger, timeToLive: null); + } + + /// + /// Registers an AI agent factory with the specified name, trigger options, and time-to-live configuration. + /// + /// If both triggers are disabled, the agent will not be accessible via HTTP or MCP tool + /// endpoints. This method can be used to register multiple agent factories with different configurations. + /// The options object to which the AI agent factory will be added. Cannot be null. + /// The unique name used to identify the AI agent factory. Cannot be null. + /// A delegate that creates an instance of the AI agent using the provided service provider. Cannot be null. + /// true to enable the HTTP trigger for the agent; otherwise, false. + /// true to enable the MCP tool trigger for the agent; otherwise, false. + /// Optional time-to-live for this agent's entities. + /// The same DurableAgentsOptions instance, allowing for method chaining. + public static DurableAgentsOptions AddAIAgentFactory( + this DurableAgentsOptions options, + string name, + Func factory, + bool enableHttpTrigger, + bool enableMcpToolTrigger, + TimeSpan? timeToLive) { ArgumentNullException.ThrowIfNull(options); ArgumentNullException.ThrowIfNull(name); @@ -122,7 +145,7 @@ public static class DurableAgentsOptionsExtensions agentOptions.HttpTrigger.IsEnabled = enableHttpTrigger; agentOptions.McpToolTrigger.IsEnabled = enableMcpToolTrigger; - options.AddAIAgentFactory(name, factory); + options.AddAIAgentFactory(name, factory, timeToLive); s_agentOptions[name] = agentOptions; return options; } diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptionsExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptionsExtensions.cs index cc7625f9e0..b85a065dc3 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptionsExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableOptionsExtensions.cs @@ -26,78 +26,90 @@ public static class DurableOptionsExtensions /// The Functions application builder for method chaining. /// /// This method provides a unified configuration point for both durable agents and workflows. - /// Agents configured here will be automatically registered when referenced in workflows. + /// It automatically generates HTTP API endpoints for agents and workflows, and configures + /// the necessary middleware and services for durable execution. /// public static FunctionsApplicationBuilder ConfigureDurableOptions( this FunctionsApplicationBuilder builder, Action configure) { + ArgumentNullException.ThrowIfNull(builder); ArgumentNullException.ThrowIfNull(configure); DurableOptions options = new(); configure(options); - // Register the unified DurableOptions for components that need both agents and workflows - builder.Services.AddSingleton(options); - - // Register AgentsOption for backward compatibility. Can be removed if needed, after syncing with team. - builder.Services.AddSingleton(options.Agents); - - // Configure agents using the existing infrastructure - builder.Services.ConfigureDurableAgents(agentOpts => - { - // Copy agent registrations from the unified options to the service-level options - foreach (KeyValuePair> agentFactory in options.Agents.GetAgentFactories()) - { - agentOpts.AddAIAgentFactory( - agentFactory.Key, - agentFactory.Value, - options.Agents.GetTimeToLive(agentFactory.Key)); - } - - // Copy TTL settings - agentOpts.DefaultTimeToLive = options.Agents.DefaultTimeToLive; - agentOpts.MinimumTimeToLiveSignalDelay = options.Agents.MinimumTimeToLiveSignalDelay; - }); - - builder.Services.TryAddSingleton(_ => - new DefaultFunctionsAgentOptionsProvider(DurableAgentsOptionsExtensions.GetAgentOptionsSnapshot())); - - builder.Services.AddSingleton(); - - // Handling of built-in function execution for Agent HTTP, MCP tool, or Entity invocations. - builder.UseWhen(static context => - string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentHttpFunctionEntryPoint, StringComparison.Ordinal) || - string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentMcpToolFunctionEntryPoint, StringComparison.Ordinal) || - string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunWorkflowOrechstrtationHttpFunctionEntryPoint, StringComparison.Ordinal) || - string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.InvokeWorkflowActivityFunctionEntryPoint, StringComparison.Ordinal) || - string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentEntityFunctionEntryPoint, StringComparison.Ordinal)); - builder.Services.AddSingleton(); - - builder.Services.AddSingleton(); - - builder.ConfigureDurableWorker().AddTasks(t => t.AddOrchestratorFunc>( - "WorkflowRunnerOrchestration", - async (tc, inputBindingData) => - { - FunctionContext? functionContext = tc.GetFunctionContext(); - if (functionContext == null) - { - throw new InvalidOperationException("FunctionContext is not available in the orchestration context."); - } - var logger = tc.CreateReplaySafeLogger("WorkflowRunnerOrchestration"); - DurableWorkflowRunner runner = functionContext.InstanceServices.GetRequiredService(); - var orchestrationResult = await runner.RunWorkflowOrchestrationAsync(tc, inputBindingData, logger); - if (logger.IsEnabled(LogLevel.Information)) - { - logger.LogInformation("Durable workflow orchestration completed. Result:{Result}", string.Join(",", orchestrationResult)); - } - - return orchestrationResult; - })); - - builder.Services.AddSingleton(); + RegisterServices(builder, options); + ConfigureAgents(builder, options); + ConfigureMiddleware(builder); + ConfigureWorkflowOrchestration(builder); return builder; } + + private static void RegisterServices(FunctionsApplicationBuilder builder, DurableOptions options) + { + builder.Services.AddSingleton(options); + builder.Services.AddSingleton(options.Agents); + builder.Services.AddSingleton(); + builder.Services.AddSingleton(); + builder.Services.AddSingleton(); + builder.Services.AddSingleton(); + + builder.Services.TryAddSingleton(_ => + new DefaultFunctionsAgentOptionsProvider(DurableAgentsOptionsExtensions.GetAgentOptionsSnapshot())); + } + + private static void ConfigureAgents(FunctionsApplicationBuilder builder, DurableOptions options) + { + builder.Services.ConfigureDurableAgents(agentOpts => + { + foreach (KeyValuePair> agentFactory in options.Agents.GetAgentFactories()) + { + bool isWorkflowOnly = options.Agents.IsWorkflowOnly(agentFactory.Key); + + agentOpts.AddAIAgentFactory( + agentFactory.Key, + agentFactory.Value, + enableHttpTrigger: !isWorkflowOnly, + enableMcpToolTrigger: false, + timeToLive: options.Agents.GetTimeToLive(agentFactory.Key)); + } + + agentOpts.DefaultTimeToLive = options.Agents.DefaultTimeToLive; + agentOpts.MinimumTimeToLiveSignalDelay = options.Agents.MinimumTimeToLiveSignalDelay; + }); + } + + private static void ConfigureMiddleware(FunctionsApplicationBuilder builder) + { + builder.UseWhen(static context => + IsBuiltInFunction(context.FunctionDefinition.EntryPoint)); + } + + private static bool IsBuiltInFunction(string? entryPoint) + { + return string.Equals(entryPoint, BuiltInFunctions.RunAgentHttpFunctionEntryPoint, StringComparison.Ordinal) + || string.Equals(entryPoint, BuiltInFunctions.RunAgentMcpToolFunctionEntryPoint, StringComparison.Ordinal) + || string.Equals(entryPoint, BuiltInFunctions.RunWorkflowOrechstrtationHttpFunctionEntryPoint, StringComparison.Ordinal) + || string.Equals(entryPoint, BuiltInFunctions.InvokeWorkflowActivityFunctionEntryPoint, StringComparison.Ordinal) + || string.Equals(entryPoint, BuiltInFunctions.RunAgentEntityFunctionEntryPoint, StringComparison.Ordinal); + } + + private static void ConfigureWorkflowOrchestration(FunctionsApplicationBuilder builder) + { + builder.ConfigureDurableWorker().AddTasks(tasks => + tasks.AddOrchestratorFunc>( + "WorkflowRunnerOrchestration", + async (orchestrationContext, request) => + { + FunctionContext functionContext = orchestrationContext.GetFunctionContext() + ?? throw new InvalidOperationException("FunctionContext is not available in the orchestration context."); + + DurableWorkflowRunner runner = functionContext.InstanceServices.GetRequiredService(); + ILogger logger = orchestrationContext.CreateReplaySafeLogger("WorkflowRunnerOrchestration"); + + return await runner.RunWorkflowOrchestrationAsync(orchestrationContext, request, logger).ConfigureAwait(true); + })); + } } diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowFunctionMetadataTransformer.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowFunctionMetadataTransformer.cs index 5d02405ea5..4f529eb54d 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowFunctionMetadataTransformer.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowFunctionMetadataTransformer.cs @@ -29,6 +29,9 @@ internal sealed class DurableWorkflowFunctionMetadataTransformer : IFunctionMeta { this._logger.LogAddingWorkflowFunction(workflow.Key); + // Currently due to how durable executor is registered, we are not able to bind TaskOrechestrationContext parameter properly + // because the InputBinding for TOC happens inside the DurableExecutor (rathen than in an input converter). + // So for now, we are going to use single orchestration function for all workflows. //original.Add(CreateOrchestrationTrigger(workflow.Key)); // We also want to create an HTTP trigger for this orchestration so users can start it via HTTP. @@ -58,14 +61,13 @@ internal sealed class DurableWorkflowFunctionMetadataTransformer : IFunctionMeta { if (executorInfos.TryGetValue(executorId, out ExecutorInfo? executorInfo)) { - // string functionName = $"{AgentSessionId.ToEntityName(workflow.Key)}-{executorId}"; //.Split("_")[0] - string functionName = $"{AgentSessionId.ToEntityName(workflow.Key)}-{executorId.Split("_")[0]}"; //.Split("_")[0] + string functionName = $"dafx-{executorId.Split("_")[0]}"; // Check if the executor type is an agent-related type if (IsAgentExecutorType(executorInfo.ExecutorType)) { this._logger.LogAddingAgentEntityFunction(executorId, executorInfo.ExecutorType.TypeName, workflow.Key); - original.Add(CreateAgentTrigger(functionName)); + //original.Add(CreateAgentTrigger(functionName)); } else { @@ -140,19 +142,19 @@ internal sealed class DurableWorkflowFunctionMetadataTransformer : IFunctionMeta }; } - private static DefaultFunctionMetadata CreateAgentTrigger(string functionName) - { - return new DefaultFunctionMetadata() - { - Name = functionName, - Language = "dotnet-isolated", - RawBindings = - [ - """{"name":"encodedEntityRequest","type":"entityTrigger","direction":"In"}""", - """{"name":"client","type":"durableClient","direction":"In"}""" - ], - EntryPoint = BuiltInFunctions.RunAgentEntityFunctionEntryPoint, - ScriptFile = BuiltInFunctions.ScriptFile, - }; - } + //private static DefaultFunctionMetadata CreateAgentTrigger(string functionName) + //{ + // return new DefaultFunctionMetadata() + // { + // Name = functionName, + // Language = "dotnet-isolated", + // RawBindings = + // [ + // """{"name":"encodedEntityRequest","type":"entityTrigger","direction":"In"}""", + // """{"name":"client","type":"durableClient","direction":"In"}""" + // ], + // EntryPoint = BuiltInFunctions.RunAgentEntityFunctionEntryPoint, + // ScriptFile = BuiltInFunctions.ScriptFile, + // }; + //} } diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs index 5afa6778a0..537ee961cc 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs @@ -1,5 +1,6 @@ // Copyright (c) Microsoft. All rights reserved. +using System.Text.Json; using Microsoft.Agents.AI.DurableTask; using Microsoft.Agents.AI.Workflows; using Microsoft.Agents.AI.Workflows.Checkpointing; @@ -9,181 +10,312 @@ using Microsoft.Extensions.Logging; namespace Microsoft.Agents.AI.Hosting.AzureFunctions; +/// +/// Executes workflow orchestrations and activity functions for durable workflows. +/// internal sealed class DurableWorkflowRunner { private readonly DurableWorkflowOptions _options; private readonly ILogger _logger; + /// + /// Initializes a new instance of the class. + /// + /// The logger instance. + /// The durable options containing workflow configurations. public DurableWorkflowRunner(ILogger logger, DurableOptions durableOptions) { - this._logger = logger; + ArgumentNullException.ThrowIfNull(logger); ArgumentNullException.ThrowIfNull(durableOptions); + + this._logger = logger; this._options = durableOptions.Workflows; } - internal async Task> RunWorkflowOrchestrationAsync(TaskOrchestrationContext taskOrchestrationContext, DuableWorkflowRunRequest input, ILogger logger) + /// + /// Runs a workflow orchestration. + /// + /// The task orchestration context. + /// The workflow run request containing workflow name and input. + /// The replay-safe logger for orchestration logging. + /// A list containing the workflow execution result. + internal async Task> RunWorkflowOrchestrationAsync( + TaskOrchestrationContext context, + DuableWorkflowRunRequest request, + ILogger logger) { - ArgumentNullException.ThrowIfNull(taskOrchestrationContext); - ArgumentNullException.ThrowIfNull(input); + ArgumentNullException.ThrowIfNull(context); + ArgumentNullException.ThrowIfNull(request); - string workflowName = input.WorkflowName; - - logger.LogAttemptingToRunWorkflow(workflowName); - - if (!this._options.Workflows.TryGetValue(workflowName, out Workflow? wf)) + if (!this._options.Workflows.TryGetValue(request.WorkflowName, out Workflow? workflow)) { - throw new InvalidOperationException($"Workflow '{workflowName}' not found."); + throw new InvalidOperationException($"Workflow '{request.WorkflowName}' not found."); } - logger.LogRunningWorkflow(wf.Name); - - var result = await this.RunExecutorsInWorkFlowAsync(taskOrchestrationContext, wf, input.Input, logger); + logger.LogRunningWorkflow(workflow.Name); + string result = await this.ExecuteWorkflowLevelsAsync(context, workflow, request.Input, logger).ConfigureAwait(true); return [result]; } - private async Task RunExecutorsInWorkFlowAsync( - TaskOrchestrationContext taskOrchestrationContext, - Workflow workflow, - string initialInput, - ILogger logger) - { - List executorResult = []; - - if (logger.IsEnabled(LogLevel.Information)) - { - foreach (WorkflowExecutorInfo executorInfo in WorkflowHelper.GetExecutorsFromWorkflowInOrder(workflow)) - { - string triggerName = this.BuildTriggerName(workflow.Name!, executorInfo.ExecutorId); - logger.LogInformation( - " Scheduling executor '{ExecutorId}' (IsAgentic: {IsAgentic}) with trigger name '{TriggerName}'", - executorInfo.ExecutorId, - executorInfo.IsAgenticExecutor, - triggerName); - - string input = executorResult.Count == 0 ? initialInput : executorResult.Last()!; - if (!executorInfo.IsAgenticExecutor) - { - var result = await taskOrchestrationContext.CallActivityAsync(triggerName, input); - executorResult.Add(result); - } - else - { - string AgentName = this.GetAgentNameFromExecutorId(workflow.Name!, executorInfo.ExecutorId); - logger.LogInformation( - " Invoking agentic executor '{ExecutorId}'", - AgentName); - DurableAIAgent agent = taskOrchestrationContext.GetAgent(AgentName); - if (agent != null) - { - AgentThread destinationThread = agent.GetNewThread(); - var agentResponse = await agent.RunAsync(input, destinationThread); - executorResult.Add(agentResponse.Text); - logger.LogInformation( - "Agentic executor '{ExecutorId}' completed with response: {AgentResponse}", - AgentName, - agentResponse); - } - } - } - - // return final result - return executorResult.Last(); - } - - return string.Empty; - } - - private string GetAgentNameFromExecutorId(string workflowName, string executorId) - { - // Example: "InspirationBot_edaac621050849efb1a62805fa03d3f8" - var parts = executorId.Split('_'); - return parts.Length > 0 ? parts[0] : executorId; - } - - private string BuildTriggerName(string workflowName, string executorName) - { - return $"dafx-{workflowName}-{executorName}"; - } - - internal async Task ExecuteActivityAsync(string activityFunctionName, string input, FunctionContext functionContext) + /// + /// Executes an activity function for a workflow executor. + /// + /// The name of the activity function to execute. + /// The input string for the executor. + /// The Azure Functions context. + /// The serialized result of the executor. + internal async Task ExecuteActivityAsync( + string activityFunctionName, + string input, + FunctionContext functionContext) { ArgumentNullException.ThrowIfNull(activityFunctionName); ArgumentNullException.ThrowIfNull(input); ArgumentNullException.ThrowIfNull(functionContext); - // Parse the activity function name to extract workflow name and executor name - // Format: "dafx-{workflowName}-{executorName}" - if (!activityFunctionName.StartsWith("dafx-", StringComparison.Ordinal)) + string executorName = ParseExecutorName(activityFunctionName); + + if (!this._options.Executors.TryGetExecutor(executorName, out ExecutorRegistration? registration) || registration is null) { - throw new InvalidOperationException($"Activity function name '{activityFunctionName}' does not start with 'dafx-' prefix."); + throw new InvalidOperationException($"Executor '{executorName}' not found in the executor registry."); } - string nameWithoutPrefix = activityFunctionName["dafx-".Length..]; - string[] parts = nameWithoutPrefix.Split('-', 2); + this._logger.LogExecutingActivity(registration.ExecutorId, executorName); - if (parts.Length != 2) + Executor executor = await registration.CreateExecutorInstanceAsync("activity-run", CancellationToken.None) + .ConfigureAwait(false); + + Type inputType = executor.InputTypes.FirstOrDefault() ?? typeof(string); + object typedInput = DeserializeInput(input, inputType); + + object? result = await executor.ExecuteAsync( + typedInput, + new TypeId(inputType), + new MinimalActivityContext(registration.ExecutorId), + CancellationToken.None).ConfigureAwait(false); + + return SerializeResult(result); + } + + private async Task ExecuteWorkflowLevelsAsync( + TaskOrchestrationContext context, + Workflow workflow, + string initialInput, + ILogger logger) + { + WorkflowExecutionPlan plan = WorkflowHelper.GetExecutionPlan(workflow); + Dictionary results = []; + + foreach (WorkflowExecutionLevel level in plan.Levels) { - throw new InvalidOperationException($"Activity function name '{activityFunctionName}' is not in the expected format 'dafx-{{workflowName}}-{{executorName}}'."); + if (level.Executors.Count == 1) + { + WorkflowExecutorInfo executorInfo = level.Executors[0]; + string input = GetExecutorInput(executorInfo.ExecutorId, initialInput, results, plan); + results[executorInfo.ExecutorId] = await this.ExecuteExecutorAsync(context, workflow, executorInfo, input, logger).ConfigureAwait(true); + } + else + { + List> tasks = []; + foreach (WorkflowExecutorInfo executorInfo in level.Executors) + { + string input = GetExecutorInput(executorInfo.ExecutorId, initialInput, results, plan); + tasks.Add(this.ExecuteExecutorWithIdAsync(context, workflow, executorInfo, input, logger)); + } + + foreach ((string id, string result) in await Task.WhenAll(tasks).ConfigureAwait(true)) + { + results[id] = result; + } + } } - string workflowName = parts[0]; - string executorName = parts[1]; + return GetFinalResult(plan, results); + } - this._logger.LogAttemptingToExecuteActivity(workflowName, executorName); + private async Task<(string Id, string Result)> ExecuteExecutorWithIdAsync( + TaskOrchestrationContext context, + Workflow workflow, + WorkflowExecutorInfo executorInfo, + string input, + ILogger logger) + { + string result = await this.ExecuteExecutorAsync(context, workflow, executorInfo, input, logger).ConfigureAwait(true); + return (executorInfo.ExecutorId, result); + } - // Get the workflow - if (!this._options.Workflows.TryGetValue(workflowName, out Workflow? workflow)) + private async Task ExecuteExecutorAsync( + TaskOrchestrationContext context, + Workflow workflow, + WorkflowExecutorInfo executorInfo, + string input, + ILogger logger) + { + if (!executorInfo.IsAgenticExecutor) { - throw new InvalidOperationException($"Workflow '{workflowName}' not found."); + string triggerName = $"dafx-{GetBaseName(executorInfo.ExecutorId)}"; + return await context.CallActivityAsync(triggerName, input).ConfigureAwait(true); } - // Get the executor info - Dictionary executorInfos = workflow.ReflectExecutors(); + return await this.ExecuteAgentAsync(context, executorInfo, input, logger).ConfigureAwait(true); + } - // Find the executor by matching the executor name (which may have a GUID suffix) - KeyValuePair executorPair = executorInfos.FirstOrDefault(e => - e.Key.StartsWith(executorName + "_", StringComparison.Ordinal) || - string.Equals(e.Key, executorName, StringComparison.Ordinal)); + private async Task ExecuteAgentAsync( + TaskOrchestrationContext context, + WorkflowExecutorInfo executorInfo, + string input, + ILogger logger) + { + string agentName = GetBaseName(executorInfo.ExecutorId); + DurableAIAgent agent = context.GetAgent(agentName); - if (executorPair.Key == null) + if (agent is null) { - throw new InvalidOperationException($"Executor '{executorName}' not found in workflow '{workflowName}'."); + logger.LogWarning("Agent '{AgentName}' not found", agentName); + return $"Agent '{agentName}' not found"; } - this._logger.LogExecutingActivity(executorPair.Key, executorPair.Value.ExecutorType.TypeName); + AgentThread thread = agent.GetNewThread(); + AgentRunResponse response = await agent.RunAsync(input, thread).ConfigureAwait(true); + return response.Text; + } - // Attempt to invoke the executor using Executor.ExecuteAsync - // This allows the executor to handle its own execution logic - try + private static string GetExecutorInput( + string executorId, + string initialInput, + Dictionary results, + WorkflowExecutionPlan plan) + { + List predecessors = plan.Predecessors[executorId]; + + if (predecessors.Count == 0) { - // Create the executor instance - Executor executor = await workflow.CreateExecutorInstanceAsync( - executorPair.Key, - "activity-run", - CancellationToken.None).ConfigureAwait(false); - - // Create a minimal workflow taskOrchestrationContext for the executor - MinimalActivityContext context = new(executorPair.Key); - - // Execute the executor with the input - // The executor handles its own routing logic internally - object? result = await executor.ExecuteAsync( - input, - new TypeId(typeof(string)), - context, - CancellationToken.None).ConfigureAwait(false); - - // Convert result to string - string resultString = result?.ToString() ?? string.Empty; - - this._logger.LogActivityExecuted(executorPair.Key, resultString); - return resultString; + return initialInput; } - catch (Exception ex) + + if (predecessors.Count == 1) { - this._logger.LogError(ex, "Error executing executor '{ExecutorId}' in activity", executorPair.Key); - throw; + return results.TryGetValue(predecessors[0], out string? result) ? result : initialInput; } + + List aggregated = []; + foreach (string predecessorId in predecessors) + { + if (results.TryGetValue(predecessorId, out string? result)) + { + aggregated.Add(result); + } + } + + return SerializeToJson(aggregated); + } + + private static string GetFinalResult(WorkflowExecutionPlan plan, Dictionary results) + { + WorkflowExecutionLevel lastLevel = plan.Levels[^1]; + + if (lastLevel.Executors.Count == 1) + { + return results[lastLevel.Executors[0].ExecutorId]; + } + + List finalResults = []; + foreach (WorkflowExecutorInfo executor in lastLevel.Executors) + { + if (results.TryGetValue(executor.ExecutorId, out string? result)) + { + finalResults.Add(result); + } + } + + return string.Join("\n---\n", finalResults); + } + + private static string ParseExecutorName(string activityFunctionName) + { + const string Prefix = "dafx-"; + + if (!activityFunctionName.StartsWith(Prefix, StringComparison.Ordinal)) + { + throw new InvalidOperationException( + $"Activity function name '{activityFunctionName}' does not start with '{Prefix}' prefix."); + } + + string executorName = activityFunctionName[Prefix.Length..]; + + if (string.IsNullOrEmpty(executorName)) + { + throw new InvalidOperationException( + $"Activity function name '{activityFunctionName}' is not in the expected format '{Prefix}{{executorName}}'."); + } + + return executorName; + } + + private static string GetBaseName(string executorId) + { + int underscoreIndex = executorId.IndexOf('_'); + return underscoreIndex > 0 ? executorId[..underscoreIndex] : executorId; + } + + [System.Diagnostics.CodeAnalysis.UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing known types.")] + [System.Diagnostics.CodeAnalysis.UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing known types.")] + private static string SerializeToJson(List values) + { + return JsonSerializer.Serialize(values); + } + + [System.Diagnostics.CodeAnalysis.UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing workflow types registered at startup.")] + [System.Diagnostics.CodeAnalysis.UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing workflow types registered at startup.")] + private static string SerializeResult(object? result) + { + if (result is null) + { + return string.Empty; + } + + if (result is string str) + { + return str; + } + + Type resultType = result.GetType(); + if (resultType.IsPrimitive || resultType == typeof(decimal)) + { + return result.ToString() ?? string.Empty; + } + + return JsonSerializer.Serialize(result, resultType); + } + + [System.Diagnostics.CodeAnalysis.UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")] + [System.Diagnostics.CodeAnalysis.UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")] + private static object DeserializeInput(string input, Type targetType) + { + if (targetType == typeof(string)) + { + return input; + } + + string json = input; + if (input.StartsWith('"') && input.EndsWith('"')) + { + try + { + string? innerJson = JsonSerializer.Deserialize(input); + if (innerJson is not null) + { + json = innerJson; + } + } + catch (JsonException) + { + // Not double-serialized, use original + } + } + + return JsonSerializer.Deserialize(json, targetType) + ?? throw new InvalidOperationException($"Failed to deserialize input to type '{targetType.Name}'."); } } diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/WorkflowHelper.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/WorkflowHelper.cs index 0fa3e946af..2f89291675 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/WorkflowHelper.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/WorkflowHelper.cs @@ -12,6 +12,46 @@ namespace Microsoft.Agents.AI.Hosting.AzureFunctions; /// Indicates whether this executor is an agentic executor. internal sealed record WorkflowExecutorInfo(string ExecutorId, bool IsAgenticExecutor); +/// +/// Represents a level of executors that can be executed in parallel (Fan-Out). +/// All executors in the same level have their dependencies satisfied by previous levels. +/// +/// The level number (0-based, starting from the root executor). +/// The executors that can run in parallel at this level. +/// Indicates if this level is a Fan-In point (has executors with multiple predecessors). +internal sealed record WorkflowExecutionLevel(int Level, List Executors, bool IsFanIn); + +/// +/// Represents the complete execution plan for a workflow, including parallel execution levels. +/// +internal sealed class WorkflowExecutionPlan +{ + /// + /// The execution levels in order. Each level contains executors that can run in parallel. + /// + public List Levels { get; } = []; + + /// + /// Maps each executor ID to its predecessors (for Fan-In result aggregation). + /// + public Dictionary> Predecessors { get; } = []; + + /// + /// Maps each executor ID to its successors (for Fan-Out result distribution). + /// + public Dictionary> Successors { get; } = []; + + /// + /// Gets whether this workflow has any parallel execution opportunities. + /// + public bool HasParallelism => this.Levels.Any(l => l.Executors.Count > 1); + + /// + /// Gets whether this workflow has any Fan-In points. + /// + public bool HasFanIn => this.Levels.Any(l => l.IsFanIn); +} + internal static class WorkflowHelper { /// @@ -20,20 +60,45 @@ internal static class WorkflowHelper /// The workflow instance to analyze. /// A list of executor information in topological order (execution order). public static List GetExecutorsFromWorkflowInOrder(Workflow workflow) + { + WorkflowExecutionPlan plan = GetExecutionPlan(workflow); + + // Flatten the levels into a single list for backward compatibility + List result = []; + foreach (WorkflowExecutionLevel level in plan.Levels) + { + result.AddRange(level.Executors); + } + + return result; + } + + /// + /// Analyzes the workflow and returns an execution plan that supports Fan-Out/Fan-In patterns. + /// Executors at the same level can be executed in parallel (Fan-Out). + /// Fan-In points are identified where multiple executors converge. + /// + /// The workflow instance to analyze. + /// An execution plan with parallel execution levels. + public static WorkflowExecutionPlan GetExecutionPlan(Workflow workflow) { ArgumentNullException.ThrowIfNull(workflow); Dictionary executors = workflow.ReflectExecutors(); Dictionary> edges = workflow.ReflectEdges(); - // Build adjacency list and in-degree map - Dictionary> adjacencyList = new(); - Dictionary inDegree = new(); + WorkflowExecutionPlan plan = new(); - // Initialize all executors with in-degree 0 + // Build adjacency lists (successors and predecessors) + Dictionary> successors = []; + Dictionary> predecessors = []; + Dictionary inDegree = []; + + // Initialize all executors foreach (string executorId in executors.Keys) { - adjacencyList[executorId] = new List(); + successors[executorId] = []; + predecessors[executorId] = []; inDegree[executorId] = 0; } @@ -44,80 +109,89 @@ internal static class WorkflowHelper foreach (EdgeInfo edge in edgeGroup.Value) { - // For each sink (target) in this edge foreach (string sinkId in edge.Connection.SinkIds) { - // Add edge from source to sink - adjacencyList[sourceId].Add(sinkId); - - // Increment in-degree of the sink - if (inDegree.TryGetValue(sinkId, out int currentDegree)) + if (executors.ContainsKey(sinkId)) { - inDegree[sinkId] = currentDegree + 1; + successors[sourceId].Add(sinkId); + predecessors[sinkId].Add(sourceId); + inDegree[sinkId]++; } } } } - // Perform topological sort using Kahn's algorithm - List orderedExecutorIds = new(); - Queue queue = new(); - - // Start with the workflow's starting executor - queue.Enqueue(workflow.StartExecutorId); - - // Also add any other executors with in-degree 0 (shouldn't be any if workflow is well-formed) - foreach (KeyValuePair kvp in inDegree) + // Store the graph structure in the plan + foreach (string executorId in executors.Keys) { - if (kvp.Value == 0 && kvp.Key != workflow.StartExecutorId) - { - queue.Enqueue(kvp.Key); - } + plan.Predecessors[executorId] = [.. predecessors[executorId]]; + plan.Successors[executorId] = [.. successors[executorId]]; } - while (queue.Count > 0) + // 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 processed = []; + Dictionary currentInDegree = new(inDegree); + int levelNumber = 0; + + while (processed.Count < executors.Count) { - string current = queue.Dequeue(); - orderedExecutorIds.Add(current); + // Find all executors that can be executed at this level (in-degree == 0 and not yet processed) + List currentLevelIds = []; - // For each neighbor of the current executor - foreach (string neighbor in adjacencyList[current]) + foreach (KeyValuePair kvp in currentInDegree) { - inDegree[neighbor]--; - - // If in-degree becomes 0, add to queue - if (inDegree[neighbor] == 0) + if (kvp.Value == 0 && !processed.Contains(kvp.Key)) { - queue.Enqueue(neighbor); + currentLevelIds.Add(kvp.Key); } } - } - // If result doesn't contain all executors, there might be a cycle or disconnected components - if (orderedExecutorIds.Count != executors.Count) - { - // Add any remaining executors that weren't reached - foreach (string executorId in executors.Keys) + // If no executors found but not all processed, there might be a cycle + if (currentLevelIds.Count == 0) { - if (!orderedExecutorIds.Contains(executorId)) + // Add remaining unprocessed executors + foreach (string executorId in executors.Keys) { - orderedExecutorIds.Add(executorId); + if (!processed.Contains(executorId)) + { + currentLevelIds.Add(executorId); + } + } + + if (currentLevelIds.Count == 0) + { + break; } } - } - // Convert to WorkflowExecutorInfo with agentic executor detection - List result = new(); - foreach (string executorId in orderedExecutorIds) - { - if (executors.TryGetValue(executorId, out ExecutorInfo? executorInfo)) + // 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 levelExecutors = []; + foreach (string executorId in currentLevelIds) { - bool isAgentic = IsAgentExecutorType(executorInfo.ExecutorType); - result.Add(new WorkflowExecutorInfo(executorId, isAgentic)); + processed.Add(executorId); + + if (executors.TryGetValue(executorId, out ExecutorInfo? executorInfo)) + { + bool isAgentic = IsAgentExecutorType(executorInfo.ExecutorType); + levelExecutors.Add(new WorkflowExecutorInfo(executorId, isAgentic)); + } + + // Decrement in-degree of all successors + foreach (string successor in successors[executorId]) + { + currentInDegree[successor]--; + } } + + plan.Levels.Add(new WorkflowExecutionLevel(levelNumber, levelExecutors, isFanIn)); + levelNumber++; } - return result; + return plan; } ///