diff --git a/dotnet/agent-framework-dotnet.slnx b/dotnet/agent-framework-dotnet.slnx index eafabd7c28..0690a5ef7e 100644 --- a/dotnet/agent-framework-dotnet.slnx +++ b/dotnet/agent-framework-dotnet.slnx @@ -37,6 +37,7 @@ + diff --git a/dotnet/samples/AzureFunctions/11_WorkflowSharedState/11_WorkflowSharedState.csproj b/dotnet/samples/AzureFunctions/11_WorkflowSharedState/11_WorkflowSharedState.csproj index ec1dca7683..4872f05930 100644 --- a/dotnet/samples/AzureFunctions/11_WorkflowSharedState/11_WorkflowSharedState.csproj +++ b/dotnet/samples/AzureFunctions/11_WorkflowSharedState/11_WorkflowSharedState.csproj @@ -1,4 +1,4 @@ - + net10.0 v4 @@ -14,6 +14,10 @@ + + + + diff --git a/dotnet/samples/AzureFunctions/12_ConditionalEdges/12_ConditionalEdges.csproj b/dotnet/samples/AzureFunctions/12_ConditionalEdges/12_ConditionalEdges.csproj new file mode 100644 index 0000000000..14ce86c2b9 --- /dev/null +++ b/dotnet/samples/AzureFunctions/12_ConditionalEdges/12_ConditionalEdges.csproj @@ -0,0 +1,48 @@ + + + net10.0 + v4 + Exe + enable + enable + + SingleAgent + SingleAgent + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/AzureFunctions/12_ConditionalEdges/OrderIdParserExecutor.cs b/dotnet/samples/AzureFunctions/12_ConditionalEdges/OrderIdParserExecutor.cs new file mode 100644 index 0000000000..31ad643fda --- /dev/null +++ b/dotnet/samples/AzureFunctions/12_ConditionalEdges/OrderIdParserExecutor.cs @@ -0,0 +1,98 @@ +// Copyright (c) Microsoft. All rights reserved. + +// This sample demonstrates how to use durable state management in Azure Functions workflows. +// The OrderIdParserExecutor writes a value to shared state, and the FraudValidation reads it back. +// The state is persisted durably using Durable Entities behind the scenes. + +using Microsoft.Agents.AI.Workflows; + +namespace SingleAgent; + +/// +/// Constants for shared state scopes used across executors. +/// +internal static class SharedStateConstants +{ + public const string MessageScope = "MessageState"; + public const string ProcessedMessageKey = "ProcessedMessage"; +} + +internal sealed class Order +{ + public Order(string id, decimal amount) + { + this.Id = id; + this.Amount = amount; + } + public string Id { get; } + public decimal Amount { get; } + public Customer? Customer { get; set; } + public string? PaymentReferenceNumber { get; set; } +} + +public sealed record Customer(int Id, string Name, bool IsBlocked); + +internal sealed class OrderIdParserExecutor() : Executor("OrderIdParserExecutor") +{ + public override async ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + return GetOrder(message); + } + + private static Order GetOrder(string id) + { + // Simulate fetching order details + return new Order(id, 100.0m); + } +} + +internal sealed class OrderEnrich() : Executor("EnrichOrder") +{ + public override async ValueTask HandleAsync(Order message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + message.Customer = GetCustomerForOrder(message.Id); + return message; + } + + private static Customer GetCustomerForOrder(string orderId) + { + if (orderId.Contains('B')) + { + return new Customer(101, "George", true); + } + + return new Customer(201, "Jerry", false); + } +} + +internal sealed class PaymentProcesserExecutor() : Executor("PaymentProcesserExecutor") +{ + public override async ValueTask HandleAsync(Order message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // Call payment gateway. + message.PaymentReferenceNumber = Guid.NewGuid().ToString().Substring(0, 4); + return message; + } +} + +internal sealed class NotifyFraudExecutor() : Executor("NotifyFraud") +{ + public override async ValueTask HandleAsync(Order message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // Notify fraud team. + return $"Order {message.Id} flagged as fraudulent for customer {message.Customer?.Name}."; + } +} + +internal static class OrderRouteConditions +{ + /// + /// Returns a condition that evaluates to true when the customer is blocked. + /// + internal static Func WhenBlocked() => order => order?.Customer?.IsBlocked == true; + + /// + /// Returns a condition that evaluates to true when the customer is not blocked. + /// + internal static Func WhenNotBlocked() => order => order?.Customer?.IsBlocked == false; +} diff --git a/dotnet/samples/AzureFunctions/12_ConditionalEdges/Program.cs b/dotnet/samples/AzureFunctions/12_ConditionalEdges/Program.cs new file mode 100644 index 0000000000..8cfa0b8126 --- /dev/null +++ b/dotnet/samples/AzureFunctions/12_ConditionalEdges/Program.cs @@ -0,0 +1,24 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Hosting.AzureFunctions; +using Microsoft.Agents.AI.Workflows; +using Microsoft.Azure.Functions.Worker.Builder; +using Microsoft.Extensions.Hosting; +using SingleAgent; + +OrderIdParserExecutor orderParser = new(); +OrderEnrich orderEnrich = new(); +PaymentProcesserExecutor paymentProcessor = new(); +NotifyFraudExecutor notifyFraud = new(); + +WorkflowBuilder builder = new(orderParser); +builder.AddEdge(orderParser, orderEnrich); +builder.AddEdge(orderEnrich, notifyFraud, condition: OrderRouteConditions.WhenBlocked()); +builder.AddEdge(orderEnrich, paymentProcessor, condition: OrderRouteConditions.WhenNotBlocked()); + +var workflow = builder.WithName("ProcessOrder").Build(); + +FunctionsApplication.CreateBuilder(args) + .ConfigureFunctionsWebApplication() + .ConfigureDurableOptions(options => options.Workflows.AddWorkflow(workflow)) + .Build().Run(); diff --git a/dotnet/samples/AzureFunctions/12_ConditionalEdges/README.md b/dotnet/samples/AzureFunctions/12_ConditionalEdges/README.md new file mode 100644 index 0000000000..d4ac968978 --- /dev/null +++ b/dotnet/samples/AzureFunctions/12_ConditionalEdges/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/12_ConditionalEdges/demo.http b/dotnet/samples/AzureFunctions/12_ConditionalEdges/demo.http new file mode 100644 index 0000000000..2b193a6f6f --- /dev/null +++ b/dotnet/samples/AzureFunctions/12_ConditionalEdges/demo.http @@ -0,0 +1,14 @@ +# Default endpoint address for local testing +@authority=http://localhost:7071 + +### Start the workflow +POST {{authority}}/api/workflows/ProcessOrder/run +Content-Type: text/plain + +B123 + +### Start second workflow +POST {{authority}}/api/workflows/ProcessOrder/run +Content-Type: text/plain + +456 \ No newline at end of file diff --git a/dotnet/samples/AzureFunctions/12_ConditionalEdges/host.json b/dotnet/samples/AzureFunctions/12_ConditionalEdges/host.json new file mode 100644 index 0000000000..9384a0a583 --- /dev/null +++ b/dotnet/samples/AzureFunctions/12_ConditionalEdges/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/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs index 5a966b4688..0144ce83d5 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs @@ -187,16 +187,25 @@ public class DurableWorkflowRunner foreach (WorkflowExecutionLevel level in plan.Levels) { - if (level.Executors.Count == 1) + // Filter executors based on edge conditions from their predecessors + List eligibleExecutors = GetEligibleExecutors(level.Executors, results, plan, logger); + + if (eligibleExecutors.Count == 0) { - WorkflowExecutorInfo executorInfo = level.Executors[0]; + // No eligible executors at this level, continue to next level + continue; + } + + if (eligibleExecutors.Count == 1) + { + WorkflowExecutorInfo executorInfo = eligibleExecutors[0]; string input = GetExecutorInput(executorInfo.ExecutorId, initialInput, results, plan); results[executorInfo.ExecutorId] = await this.ExecuteExecutorAsync(context, executorInfo, input, logger).ConfigureAwait(true); } else { List> tasks = []; - foreach (WorkflowExecutorInfo executorInfo in level.Executors) + foreach (WorkflowExecutorInfo executorInfo in eligibleExecutors) { string input = GetExecutorInput(executorInfo.ExecutorId, initialInput, results, plan); tasks.Add(this.ExecuteExecutorWithIdAsync(context, executorInfo, input, logger)); @@ -212,6 +221,115 @@ public class DurableWorkflowRunner return GetFinalResult(plan, results); } + /// + /// Filters executors based on their incoming edge conditions. + /// An executor is eligible if all its incoming edges have conditions that evaluate to true, + /// or if the edges have no conditions. + /// + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")] + private static List GetEligibleExecutors( + List executors, + Dictionary results, + WorkflowExecutionPlan plan, + ILogger logger) + { + List eligible = []; + + foreach (WorkflowExecutorInfo executorInfo in executors) + { + List predecessors = plan.Predecessors[executorInfo.ExecutorId]; + + // Root executor (no predecessors) is always eligible + if (predecessors.Count == 0) + { + eligible.Add(executorInfo); + continue; + } + + // Check if any predecessor's edge condition allows this executor to run + bool isEligible = false; + foreach (string predecessorId in predecessors) + { + // Get the condition for this edge (predecessor -> current executor) + if (!plan.EdgeConditions.TryGetValue((predecessorId, executorInfo.ExecutorId), out Func? condition)) + { + // No condition registered for this edge, assume it's eligible + isEligible = true; + break; + } + + if (condition is null) + { + // Edge has no condition, always eligible + isEligible = true; + break; + } + + // Evaluate the condition using the predecessor's result + if (results.TryGetValue(predecessorId, out string? predecessorResult)) + { + try + { + // Get the predecessor's output type for proper deserialization + Type? predecessorOutputType = plan.ExecutorOutputTypes.GetValueOrDefault(predecessorId); + + // Deserialize the predecessor result to the expected type for condition evaluation + object? resultObject = DeserializeForCondition(predecessorResult, predecessorOutputType); + if (condition(resultObject)) + { + isEligible = true; + break; + } + } + catch (Exception ex) + { + logger.LogWarning(ex, "Failed to evaluate condition for edge from '{PredecessorId}' to '{ExecutorId}'", predecessorId, executorInfo.ExecutorId); + } + } + } + + if (isEligible) + { + eligible.Add(executorInfo); + } + else + { + logger.LogExecutorSkipped(executorInfo.ExecutorId); + } + } + + return eligible; + } + + /// + /// Deserializes a JSON string result into an object for condition evaluation. + /// + [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")] + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")] + private static object? DeserializeForCondition(string json, Type? targetType) + { + if (string.IsNullOrEmpty(json)) + { + return null; + } + + try + { + if (targetType is null) + { + return JsonSerializer.Deserialize(json); + } + + return JsonSerializer.Deserialize(json, targetType); + } + catch (JsonException) + { + // If it's not valid JSON, return the string as-is + return json; + } + } + private async Task<(string Id, string Result)> ExecuteExecutorWithIdAsync( TaskOrchestrationContext context, WorkflowExecutorInfo executorInfo, diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Logs.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Logs.cs index 86794199d4..1667095a96 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Logs.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Logs.cs @@ -130,4 +130,10 @@ internal static partial class Logs Level = LogLevel.Information, Message = "Activity executed for executor '{ExecutorId}' with result: {Result}")] public static partial void LogActivityExecuted(this ILogger logger, string executorId, string result); + + [LoggerMessage( + EventId = 17, + Level = LogLevel.Information, + Message = "Executor '{ExecutorId}' skipped due to edge condition evaluation")] + public static partial void LogExecutorSkipped(this ILogger logger, string executorId); } diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs index 2a0cdc586a..a08f314433 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/WorkflowHelper.cs @@ -41,6 +41,17 @@ public sealed class WorkflowExecutionPlan /// public Dictionary> Successors { get; } = []; + /// + /// Maps edge connections (sourceId, targetId) to their condition functions. + /// The condition function takes the predecessor's result and returns true if the edge should be followed. + /// + public Dictionary<(string SourceId, string TargetId), Func?> EdgeConditions { get; } = []; + + /// + /// Maps executor IDs to their output types (for proper deserialization during condition evaluation). + /// + public Dictionary ExecutorOutputTypes { get; } = []; + /// /// Gets whether this workflow has any parallel execution opportunities. /// @@ -89,6 +100,7 @@ public static class WorkflowHelper Dictionary executors = workflow.ReflectExecutors(); Dictionary> edges = workflow.ReflectEdges(); + Dictionary<(string SourceId, string TargetId), Func?> edgeConditions = workflow.GetEdgeConditions(); WorkflowExecutionPlan plan = new(); @@ -97,12 +109,15 @@ public static class WorkflowHelper Dictionary> predecessors = []; Dictionary inDegree = []; - // Initialize all executors - foreach (string executorId in executors.Keys) + // Initialize all executors and extract their output types + foreach (KeyValuePair executor in executors) { - successors[executorId] = []; - predecessors[executorId] = []; - inDegree[executorId] = 0; + successors[executor.Key] = []; + predecessors[executor.Key] = []; + inDegree[executor.Key] = 0; + + // Extract output type from executor type (e.g., Executor -> TOutput) + plan.ExecutorOutputTypes[executor.Key] = GetExecutorOutputType(executor.Value.ExecutorType); } // Build the graph from edges @@ -124,6 +139,12 @@ public static class WorkflowHelper } } + // Store edge conditions in the plan + foreach (KeyValuePair<(string SourceId, string TargetId), Func?> condition in edgeConditions) + { + plan.EdgeConditions[condition.Key] = condition.Value; + } + // Store the graph structure in the plan foreach (string executorId in executors.Keys) { @@ -213,4 +234,41 @@ public static class WorkflowHelper return typeName.Contains("AIAgentHostExecutor", StringComparison.OrdinalIgnoreCase) && assemblyName.Contains("Microsoft.Agents.AI", StringComparison.OrdinalIgnoreCase); } + + /// + /// Extracts the output type from an executor type. + /// For Executor<TInput, TOutput>, returns TOutput. + /// For Executor<TInput>, returns null (void output). + /// + /// The executor type to analyze. + /// The output type, or null if the executor has no typed output. + private static Type? GetExecutorOutputType(Type executorType) + { + // Walk up the inheritance chain to find Executor or Executor + Type? currentType = executorType; + while (currentType is not null) + { + if (currentType.IsGenericType) + { + Type genericDefinition = currentType.GetGenericTypeDefinition(); + Type[] genericArgs = currentType.GetGenericArguments(); + + // Check for Executor (2 type parameters) + if (genericArgs.Length == 2 && genericDefinition.Name.StartsWith("Executor", StringComparison.Ordinal)) + { + return genericArgs[1]; // TOutput + } + + // Check for Executor (1 type parameter) - void return + if (genericArgs.Length == 1 && genericDefinition.Name.StartsWith("Executor", StringComparison.Ordinal)) + { + return null; + } + } + + currentType = currentType.BaseType; + } + + return null; + } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs index b0856f3839..229443638b 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Workflow.cs @@ -35,6 +35,30 @@ public class Workflow ); } + /// + /// Gets the condition functions for direct edges, keyed by (sourceId, targetId) tuple. + /// + /// A dictionary mapping edge connections to their condition functions (null if no condition). + /// This method creates a new dictionary each time it is called to ensure thread safety. + [System.Diagnostics.CodeAnalysis.SuppressMessage("Design", "CA1024:Use properties where appropriate", Justification = "Method creates a new collection on each call.")] + public Dictionary<(string SourceId, string TargetId), Func?> GetEdgeConditions() + { + Dictionary<(string SourceId, string TargetId), Func?> conditions = []; + + foreach (KeyValuePair> edgeGroup in this.Edges) + { + foreach (Edge edge in edgeGroup.Value) + { + if (edge.DirectEdgeData is DirectEdgeData directEdge) + { + conditions[(directEdge.SourceId, directEdge.SinkId)] = directEdge.Condition; + } + } + } + + return conditions; + } + /// /// Gets all executor bindings in the workflow, keyed by their ID. ///