mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Nested workflow support.
This commit is contained in:
Shyju Krishnankutty
@@ -54,6 +54,7 @@
|
||||
<Project Path="samples/Durable/Workflow/ConsoleApps/04_WorkflowAndAgents/04_WorkflowAndAgents.csproj" />
|
||||
<Project Path="samples/Durable/Workflow/ConsoleApps/05_WorkflowEvents/05_WorkflowEvents.csproj" />
|
||||
<Project Path="samples/Durable/Workflow/ConsoleApps/06_WorkflowSharedState/06_WorkflowSharedState.csproj" />
|
||||
<Project Path="samples/Durable/Workflow/ConsoleApps/07_SubWorkflows/07_SubWorkflows.csproj" />
|
||||
</Folder>
|
||||
<Folder Name="/Samples/Durable/Workflows/AzureFunctions/">
|
||||
<Project Path="samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/01_SequentialWorkflow.csproj" />
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
<PropertyGroup>
|
||||
<TargetFrameworks>net10.0</TargetFrameworks>
|
||||
<OutputType>Exe</OutputType>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<Nullable>enable</Nullable>
|
||||
<AssemblyName>SubWorkflows</AssemblyName>
|
||||
<RootNamespace>SubWorkflows</RootNamespace>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Azure.Identity" />
|
||||
<PackageReference Include="Microsoft.DurableTask.Client.AzureManaged" />
|
||||
<PackageReference Include="Microsoft.DurableTask.Worker.AzureManaged" />
|
||||
<PackageReference Include="Microsoft.Extensions.Hosting" />
|
||||
</ItemGroup>
|
||||
|
||||
<!-- Local projects that should be switched to package references when using the sample outside of this MAF repo -->
|
||||
<!--
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.Agents.AI.DurableTask" />
|
||||
<PackageReference Include="Microsoft.Agents.AI.Workflows" />
|
||||
</ItemGroup>
|
||||
-->
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\..\..\..\src\Microsoft.Agents.AI.DurableTask\Microsoft.Agents.AI.DurableTask.csproj" />
|
||||
<ProjectReference Include="..\..\..\..\..\src\Microsoft.Agents.AI.OpenAI\Microsoft.Agents.AI.OpenAI.csproj" />
|
||||
</ItemGroup>
|
||||
</Project>
|
||||
@@ -0,0 +1,233 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
using Microsoft.Agents.AI.Workflows.Checkpointing;
|
||||
|
||||
namespace SubWorkflows;
|
||||
|
||||
/// <summary>
|
||||
/// Event emitted when the fraud check risk score is calculated.
|
||||
/// </summary>
|
||||
internal sealed class FraudRiskAssessedEvent(int riskScore) : WorkflowEvent($"Risk score: {riskScore}/100")
|
||||
{
|
||||
public int RiskScore => riskScore;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Represents an order being processed through the workflow.
|
||||
/// </summary>
|
||||
internal sealed class OrderInfo
|
||||
{
|
||||
public required string OrderId { get; set; }
|
||||
|
||||
public decimal Amount { get; set; }
|
||||
|
||||
public string? PaymentTransactionId { get; set; }
|
||||
|
||||
public string? TrackingNumber { get; set; }
|
||||
|
||||
public string? Carrier { get; set; }
|
||||
}
|
||||
|
||||
// Main workflow executors
|
||||
|
||||
/// <summary>
|
||||
/// Entry point executor that receives the order ID and creates an OrderInfo object.
|
||||
/// </summary>
|
||||
internal sealed class OrderReceived() : Executor<string, OrderInfo>("OrderReceived")
|
||||
{
|
||||
public override ValueTask<OrderInfo> HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
Console.WriteLine();
|
||||
Console.ForegroundColor = ConsoleColor.Cyan;
|
||||
Console.WriteLine($"[OrderReceived] Processing order '{message}'");
|
||||
Console.ResetColor();
|
||||
|
||||
OrderInfo order = new()
|
||||
{
|
||||
OrderId = message,
|
||||
Amount = 99.99m // Simulated order amount
|
||||
};
|
||||
|
||||
return ValueTask.FromResult(order);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Final executor that outputs the completed order summary.
|
||||
/// </summary>
|
||||
internal sealed class OrderCompleted() : Executor<OrderInfo, string>("OrderCompleted")
|
||||
{
|
||||
public override ValueTask<string> HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
Console.WriteLine();
|
||||
Console.ForegroundColor = ConsoleColor.Green;
|
||||
Console.WriteLine("┌─────────────────────────────────────────────────────────────────┐");
|
||||
Console.WriteLine($"│ [OrderCompleted] Order '{message.OrderId}' successfully processed!");
|
||||
Console.WriteLine($"│ Payment: {message.PaymentTransactionId}");
|
||||
Console.WriteLine($"│ Shipping: {message.Carrier} - {message.TrackingNumber}");
|
||||
Console.WriteLine("└─────────────────────────────────────────────────────────────────┘");
|
||||
Console.ResetColor();
|
||||
|
||||
return ValueTask.FromResult($"Order {message.OrderId} completed. Tracking: {message.TrackingNumber}");
|
||||
}
|
||||
}
|
||||
|
||||
// Payment sub-workflow executors
|
||||
|
||||
/// <summary>
|
||||
/// Validates payment information for an order.
|
||||
/// </summary>
|
||||
internal sealed class ValidatePayment() : Executor<OrderInfo, OrderInfo>("ValidatePayment")
|
||||
{
|
||||
public override async ValueTask<OrderInfo> HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
Console.WriteLine();
|
||||
Console.ForegroundColor = ConsoleColor.Yellow;
|
||||
Console.WriteLine($" [Payment/ValidatePayment] Validating payment for order '{message.OrderId}'...");
|
||||
Console.ResetColor();
|
||||
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken);
|
||||
|
||||
Console.ForegroundColor = ConsoleColor.Yellow;
|
||||
Console.WriteLine($" [Payment/ValidatePayment] Payment validated for ${message.Amount}");
|
||||
Console.ResetColor();
|
||||
|
||||
return message;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Charges the payment for an order.
|
||||
/// </summary>
|
||||
internal sealed class ChargePayment() : Executor<OrderInfo, OrderInfo>("ChargePayment")
|
||||
{
|
||||
public override async ValueTask<OrderInfo> HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
Console.ForegroundColor = ConsoleColor.Yellow;
|
||||
Console.WriteLine($" [Payment/ChargePayment] Charging ${message.Amount} for order '{message.OrderId}'...");
|
||||
Console.ResetColor();
|
||||
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken);
|
||||
|
||||
message.PaymentTransactionId = $"TXN-{Guid.NewGuid().ToString("N")[..8].ToUpperInvariant()}";
|
||||
|
||||
Console.ForegroundColor = ConsoleColor.Yellow;
|
||||
Console.WriteLine($" [Payment/ChargePayment] ✓ Payment processed: {message.PaymentTransactionId}");
|
||||
Console.ResetColor();
|
||||
|
||||
return message;
|
||||
}
|
||||
}
|
||||
|
||||
// FraudCheck sub-sub-workflow executors (nested inside Payment)
|
||||
|
||||
/// <summary>
|
||||
/// Analyzes transaction patterns for potential fraud.
|
||||
/// </summary>
|
||||
internal sealed class AnalyzePatterns() : Executor<OrderInfo, OrderInfo>("AnalyzePatterns")
|
||||
{
|
||||
public override async ValueTask<OrderInfo> HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
Console.ForegroundColor = ConsoleColor.DarkYellow;
|
||||
Console.WriteLine($" [Payment/FraudCheck/AnalyzePatterns] Analyzing patterns for order '{message.OrderId}'...");
|
||||
Console.ResetColor();
|
||||
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken);
|
||||
|
||||
// Store analysis results in shared state for the next executor in this sub-workflow
|
||||
int patternsFound = new Random().Next(0, 5);
|
||||
await context.QueueStateUpdateAsync("patternsFound", patternsFound, cancellationToken: cancellationToken);
|
||||
|
||||
Console.ForegroundColor = ConsoleColor.DarkYellow;
|
||||
Console.WriteLine($" [Payment/FraudCheck/AnalyzePatterns] ✓ Pattern analysis complete ({patternsFound} suspicious patterns)");
|
||||
Console.ResetColor();
|
||||
|
||||
return message;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Calculates a risk score for the transaction.
|
||||
/// </summary>
|
||||
internal sealed class CalculateRiskScore() : Executor<OrderInfo, OrderInfo>("CalculateRiskScore")
|
||||
{
|
||||
public override async ValueTask<OrderInfo> HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
Console.ForegroundColor = ConsoleColor.DarkYellow;
|
||||
Console.WriteLine($" [Payment/FraudCheck/CalculateRiskScore] Calculating risk score for order '{message.OrderId}'...");
|
||||
Console.ResetColor();
|
||||
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken);
|
||||
|
||||
// Read the pattern count from shared state (written by AnalyzePatterns)
|
||||
int patternsFound = await context.ReadStateAsync<int>("patternsFound", cancellationToken: cancellationToken);
|
||||
int riskScore = Math.Min(patternsFound * 20 + new Random().Next(1, 20), 100);
|
||||
|
||||
// Emit a workflow event from within a nested sub-workflow
|
||||
await context.AddEventAsync(new FraudRiskAssessedEvent(riskScore), cancellationToken);
|
||||
|
||||
Console.ForegroundColor = ConsoleColor.DarkYellow;
|
||||
Console.WriteLine($" [Payment/FraudCheck/CalculateRiskScore] ✓ Risk score: {riskScore}/100 (based on {patternsFound} patterns)");
|
||||
Console.ResetColor();
|
||||
|
||||
return message;
|
||||
}
|
||||
}
|
||||
|
||||
// Shipping sub-workflow executors
|
||||
|
||||
/// <summary>
|
||||
/// Selects a shipping carrier for an order.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// This executor uses <see cref="Executor{TInput}"/> (void return) combined with
|
||||
/// <see cref="IWorkflowContext.SendMessageAsync"/> to forward the order to the next
|
||||
/// connected executor (CreateShipment). This demonstrates explicit typed message passing
|
||||
/// as an alternative to returning a value from the handler.
|
||||
/// </remarks>
|
||||
internal sealed class SelectCarrier() : Executor<OrderInfo>("SelectCarrier")
|
||||
{
|
||||
public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
Console.WriteLine();
|
||||
Console.ForegroundColor = ConsoleColor.Blue;
|
||||
Console.WriteLine($" [Shipping/SelectCarrier] Selecting carrier for order '{message.OrderId}'...");
|
||||
Console.ResetColor();
|
||||
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken);
|
||||
|
||||
message.Carrier = message.Amount > 50 ? "Express" : "Standard";
|
||||
|
||||
Console.ForegroundColor = ConsoleColor.Blue;
|
||||
Console.WriteLine($" [Shipping/SelectCarrier] ✓ Selected carrier: {message.Carrier}");
|
||||
Console.ResetColor();
|
||||
|
||||
// Use SendMessageAsync to forward the updated order to connected executors.
|
||||
// With a void-return executor, this is the mechanism for passing data downstream.
|
||||
await context.SendMessageAsync(message, cancellationToken: cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Creates shipment and generates tracking number.
|
||||
/// </summary>
|
||||
internal sealed class CreateShipment() : Executor<OrderInfo, OrderInfo>("CreateShipment")
|
||||
{
|
||||
public override async ValueTask<OrderInfo> HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
Console.ForegroundColor = ConsoleColor.Blue;
|
||||
Console.WriteLine($" [Shipping/CreateShipment] Creating shipment for order '{message.OrderId}'...");
|
||||
Console.ResetColor();
|
||||
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken);
|
||||
|
||||
message.TrackingNumber = $"TRACK-{Guid.NewGuid().ToString("N")[..10].ToUpperInvariant()}";
|
||||
|
||||
Console.ForegroundColor = ConsoleColor.Blue;
|
||||
Console.WriteLine($" [Shipping/CreateShipment] ✓ Shipment created: {message.TrackingNumber}");
|
||||
Console.ResetColor();
|
||||
|
||||
return message;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,146 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
// This sample demonstrates nested sub-workflows. A sub-workflow can act as an executor
|
||||
// within another workflow, including multi-level nesting (sub-workflow within sub-workflow).
|
||||
|
||||
using Microsoft.Agents.AI.DurableTask;
|
||||
using Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
using Microsoft.DurableTask.Client.AzureManaged;
|
||||
using Microsoft.DurableTask.Worker.AzureManaged;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using SubWorkflows;
|
||||
|
||||
// Get DTS connection string from environment variable
|
||||
string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING")
|
||||
?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None";
|
||||
|
||||
// Build the FraudCheck sub-workflow (this will be nested inside the Payment sub-workflow)
|
||||
AnalyzePatterns analyzePatterns = new();
|
||||
CalculateRiskScore calculateRiskScore = new();
|
||||
|
||||
Workflow fraudCheckWorkflow = new WorkflowBuilder(analyzePatterns)
|
||||
.WithName("SubFraudCheck")
|
||||
.WithDescription("Analyzes transaction patterns and calculates risk score")
|
||||
.AddEdge(analyzePatterns, calculateRiskScore)
|
||||
.Build();
|
||||
|
||||
// Build the Payment sub-workflow: ValidatePayment -> FraudCheck (sub-workflow) -> ChargePayment
|
||||
ValidatePayment validatePayment = new();
|
||||
ExecutorBinding fraudCheckExecutor = fraudCheckWorkflow.BindAsExecutor("FraudCheck");
|
||||
ChargePayment chargePayment = new();
|
||||
|
||||
Workflow paymentWorkflow = new WorkflowBuilder(validatePayment)
|
||||
.WithName("SubPaymentProcessing")
|
||||
.WithDescription("Validates and processes payment for an order")
|
||||
.AddEdge(validatePayment, fraudCheckExecutor)
|
||||
.AddEdge(fraudCheckExecutor, chargePayment)
|
||||
.Build();
|
||||
|
||||
// Build the Shipping sub-workflow: SelectCarrier -> CreateShipment
|
||||
SelectCarrier selectCarrier = new();
|
||||
CreateShipment createShipment = new();
|
||||
|
||||
Workflow shippingWorkflow = new WorkflowBuilder(selectCarrier)
|
||||
.WithName("SubShippingArrangement")
|
||||
.WithDescription("Selects carrier and creates shipment")
|
||||
.AddEdge(selectCarrier, createShipment)
|
||||
.Build();
|
||||
|
||||
// Build the main workflow using sub-workflows as executors
|
||||
// OrderReceived -> Payment (sub-workflow) -> Shipping (sub-workflow) -> OrderCompleted
|
||||
OrderReceived orderReceived = new();
|
||||
OrderCompleted orderCompleted = new();
|
||||
ExecutorBinding paymentExecutor = paymentWorkflow.BindAsExecutor("Payment");
|
||||
ExecutorBinding shippingExecutor = shippingWorkflow.BindAsExecutor("Shipping");
|
||||
|
||||
Workflow orderProcessingWorkflow = new WorkflowBuilder(orderReceived)
|
||||
.WithName("OrderProcessing")
|
||||
.WithDescription("Processes an order through payment and shipping")
|
||||
.AddEdge(orderReceived, paymentExecutor)
|
||||
.AddEdge(paymentExecutor, shippingExecutor)
|
||||
.AddEdge(shippingExecutor, orderCompleted)
|
||||
.Build();
|
||||
|
||||
// Configure and start the host
|
||||
// Register only the main workflow - sub-workflows are discovered automatically!
|
||||
IHost host = Host.CreateDefaultBuilder(args)
|
||||
.ConfigureLogging(logging => logging.SetMinimumLevel(LogLevel.Warning))
|
||||
.ConfigureServices(services =>
|
||||
{
|
||||
services.ConfigureDurableWorkflows(
|
||||
workflowOptions => workflowOptions.AddWorkflow(orderProcessingWorkflow),
|
||||
workerBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString),
|
||||
clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString));
|
||||
})
|
||||
.Build();
|
||||
|
||||
await host.StartAsync();
|
||||
|
||||
IWorkflowClient workflowClient = host.Services.GetRequiredService<IWorkflowClient>();
|
||||
|
||||
Console.WriteLine("Durable Sub-Workflows Sample");
|
||||
Console.WriteLine("Workflow: OrderReceived -> Payment(sub) -> Shipping(sub) -> OrderCompleted");
|
||||
Console.WriteLine(" Payment contains nested FraudCheck sub-workflow (Level 2 nesting)");
|
||||
Console.WriteLine();
|
||||
Console.WriteLine("Enter an order ID (or 'exit'):");
|
||||
|
||||
while (true)
|
||||
{
|
||||
Console.Write("> ");
|
||||
string? input = Console.ReadLine();
|
||||
if (string.IsNullOrWhiteSpace(input) || input.Equals("exit", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await StartNewWorkflowAsync(input, orderProcessingWorkflow, workflowClient);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Console.WriteLine($"Error: {ex.Message}");
|
||||
}
|
||||
|
||||
Console.WriteLine();
|
||||
}
|
||||
|
||||
await host.StopAsync();
|
||||
|
||||
// Start a new workflow using streaming to observe events (including from sub-workflows)
|
||||
static async Task StartNewWorkflowAsync(string orderId, Workflow workflow, IWorkflowClient client)
|
||||
{
|
||||
Console.WriteLine($"\nStarting order processing for '{orderId}'...");
|
||||
|
||||
IStreamingWorkflowRun run = await client.StreamAsync(workflow, orderId);
|
||||
Console.WriteLine($"Run ID: {run.RunId}");
|
||||
Console.WriteLine();
|
||||
|
||||
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
|
||||
{
|
||||
switch (evt)
|
||||
{
|
||||
// Custom event emitted from the FraudCheck sub-sub-workflow
|
||||
case FraudRiskAssessedEvent e:
|
||||
Console.ForegroundColor = ConsoleColor.DarkYellow;
|
||||
Console.WriteLine($" [Event from sub-workflow] {e.GetType().Name}: Risk score {e.RiskScore}/100");
|
||||
Console.ResetColor();
|
||||
break;
|
||||
|
||||
case DurableWorkflowCompletedEvent e:
|
||||
Console.ForegroundColor = ConsoleColor.Green;
|
||||
Console.WriteLine($"✓ Order completed: {e.Result}");
|
||||
Console.ResetColor();
|
||||
break;
|
||||
|
||||
case DurableWorkflowFailedEvent e:
|
||||
Console.ForegroundColor = ConsoleColor.Red;
|
||||
Console.WriteLine($"✗ Failed: {e.ErrorMessage}");
|
||||
Console.ResetColor();
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
# Sub-Workflows Sample (Nested Workflows)
|
||||
|
||||
This sample demonstrates how to compose complex workflows from simpler, reusable sub-workflows using the Durable Task Framework. Sub-workflows run as separate orchestration instances, providing modular design, independent checkpointing, and hierarchical visualization in the DTS dashboard.
|
||||
|
||||
## Key Concepts Demonstrated
|
||||
|
||||
- **Sub-workflows**: Using `Workflow.BindAsExecutor()` to embed a workflow as an executor in another workflow
|
||||
- **Multi-level nesting**: Sub-workflows within sub-workflows (Level 2 nesting)
|
||||
- **Automatic discovery**: Registering only the main workflow; sub-workflows are discovered automatically
|
||||
- **Failure isolation**: Each sub-workflow runs as a separate orchestration instance
|
||||
- **Hierarchical visualization**: Parent-child orchestration hierarchy visible in the DTS dashboard
|
||||
- **Event propagation**: Custom workflow events (`FraudRiskAssessedEvent`) bubble up from nested sub-workflows to the streaming client
|
||||
- **Message passing**: Using `Executor<TInput>` (void return) with `SendMessageAsync` to forward typed messages to connected executors (`SelectCarrier`)
|
||||
- **Shared state within sub-workflows**: Using `QueueStateUpdateAsync`/`ReadStateAsync` to share data between executors within a sub-workflow (`AnalyzePatterns` → `CalculateRiskScore`)
|
||||
|
||||
## Overview
|
||||
|
||||
The sample implements an order processing workflow composed of two sub-workflows, one of which contains its own nested sub-workflow:
|
||||
|
||||
```
|
||||
OrderProcessing (main workflow)
|
||||
├── OrderReceived
|
||||
├── Payment (sub-workflow)
|
||||
│ ├── ValidatePayment
|
||||
│ ├── FraudCheck (sub-sub-workflow) ← Level 2 nesting!
|
||||
│ │ ├── AnalyzePatterns
|
||||
│ │ └── CalculateRiskScore
|
||||
│ └── ChargePayment
|
||||
├── Shipping (sub-workflow)
|
||||
│ ├── SelectCarrier ← Uses SendMessageAsync (void-return executor)
|
||||
│ └── CreateShipment
|
||||
└── OrderCompleted
|
||||
```
|
||||
|
||||
| Executor | Sub-Workflow | Description |
|
||||
|----------|-------------|-------------|
|
||||
| OrderReceived | Main | Receives order ID and creates order info |
|
||||
| ValidatePayment | Payment | Validates payment information |
|
||||
| AnalyzePatterns | FraudCheck (nested in Payment) | Analyzes transaction patterns, stores results in shared state |
|
||||
| CalculateRiskScore | FraudCheck (nested in Payment) | Reads shared state, calculates risk score, emits `FraudRiskAssessedEvent` |
|
||||
| ChargePayment | Payment | Charges payment amount |
|
||||
| SelectCarrier | Shipping | Selects carrier using `SendMessageAsync` (void-return executor) |
|
||||
| CreateShipment | Shipping | Creates shipment with tracking |
|
||||
| OrderCompleted | Main | Outputs completed order summary |
|
||||
|
||||
## How Sub-Workflows Work
|
||||
|
||||
1. **Build** each sub-workflow as a standalone `Workflow` using `WorkflowBuilder`
|
||||
2. **Bind** a workflow as an executor using `workflow.BindAsExecutor("name")`
|
||||
3. **Add** the bound executor as a node in the parent workflow's graph
|
||||
4. **Register** only the top-level workflow — sub-workflows are discovered and registered automatically
|
||||
|
||||
```csharp
|
||||
// Build a sub-workflow
|
||||
Workflow fraudCheckWorkflow = new WorkflowBuilder(analyzePatterns)
|
||||
.WithName("SubFraudCheck")
|
||||
.AddEdge(analyzePatterns, calculateRiskScore)
|
||||
.Build();
|
||||
|
||||
// Nest it inside another sub-workflow using BindAsExecutor
|
||||
ExecutorBinding fraudCheckExecutor = fraudCheckWorkflow.BindAsExecutor("FraudCheck");
|
||||
|
||||
Workflow paymentWorkflow = new WorkflowBuilder(validatePayment)
|
||||
.WithName("SubPaymentProcessing")
|
||||
.AddEdge(validatePayment, fraudCheckExecutor)
|
||||
.AddEdge(fraudCheckExecutor, chargePayment)
|
||||
.Build();
|
||||
|
||||
// Use the Payment sub-workflow in the main workflow
|
||||
ExecutorBinding paymentExecutor = paymentWorkflow.BindAsExecutor("Payment");
|
||||
|
||||
Workflow mainWorkflow = new WorkflowBuilder(orderReceived)
|
||||
.AddEdge(orderReceived, paymentExecutor)
|
||||
.AddEdge(paymentExecutor, orderCompleted)
|
||||
.Build();
|
||||
```
|
||||
|
||||
## Environment Setup
|
||||
|
||||
See the [README.md](../README.md) file in the parent directory for information on configuring the environment, including how to install and run the Durable Task Scheduler.
|
||||
|
||||
## Running the Sample
|
||||
|
||||
```bash
|
||||
cd dotnet/samples/Durable/Workflow/ConsoleApps/07_SubWorkflows
|
||||
dotnet run --framework net10.0
|
||||
```
|
||||
|
||||
### Sample Output
|
||||
|
||||
```text
|
||||
Durable Sub-Workflows Sample
|
||||
Workflow: OrderReceived -> Payment(sub) -> Shipping(sub) -> OrderCompleted
|
||||
Payment contains nested FraudCheck sub-workflow (Level 2 nesting)
|
||||
|
||||
Enter an order ID (or 'exit'):
|
||||
> ORD-001
|
||||
Starting order processing for 'ORD-001'...
|
||||
Run ID: abc123...
|
||||
|
||||
[OrderReceived] Processing order 'ORD-001'
|
||||
[Payment/ValidatePayment] Validating payment for order 'ORD-001'...
|
||||
[Payment/ValidatePayment] Payment validated for $99.99
|
||||
[Payment/FraudCheck/AnalyzePatterns] Analyzing patterns for order 'ORD-001'...
|
||||
[Payment/FraudCheck/AnalyzePatterns] ✓ Pattern analysis complete (2 suspicious patterns)
|
||||
[Payment/FraudCheck/CalculateRiskScore] Calculating risk score for order 'ORD-001'...
|
||||
[Payment/FraudCheck/CalculateRiskScore] ✓ Risk score: 53/100 (based on 2 patterns)
|
||||
[Event from sub-workflow] FraudRiskAssessedEvent: Risk score 53/100
|
||||
[Payment/ChargePayment] Charging $99.99 for order 'ORD-001'...
|
||||
[Payment/ChargePayment] ✓ Payment processed: TXN-A1B2C3D4
|
||||
[Shipping/SelectCarrier] Selecting carrier for order 'ORD-001'...
|
||||
[Shipping/SelectCarrier] ✓ Selected carrier: Express
|
||||
[Shipping/CreateShipment] Creating shipment for order 'ORD-001'...
|
||||
[Shipping/CreateShipment] ✓ Shipment created: TRACK-I9J0K1L2M3
|
||||
┌─────────────────────────────────────────────────────────────────┐
|
||||
│ [OrderCompleted] Order 'ORD-001' successfully processed!
|
||||
│ Payment: TXN-A1B2C3D4
|
||||
│ Shipping: Express - TRACK-I9J0K1L2M3
|
||||
└─────────────────────────────────────────────────────────────────┘
|
||||
✓ Order completed: Order ORD-001 completed. Tracking: TRACK-I9J0K1L2M3
|
||||
|
||||
> exit
|
||||
```
|
||||
@@ -14,19 +14,20 @@ using Microsoft.Extensions.Logging;
|
||||
namespace Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
|
||||
/// <summary>
|
||||
/// Dispatches workflow executors to either activities or AI agents.
|
||||
/// Dispatches workflow executors to activities, AI agents, or sub-orchestrations.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Called during the dispatch phase of each superstep by
|
||||
/// <c>DurableWorkflowRunner.DispatchExecutorsInParallelAsync</c>. For each executor that has
|
||||
/// pending input, this dispatcher determines whether the executor is an AI agent (stateful,
|
||||
/// backed by Durable Entities) or a regular activity, and invokes the appropriate Durable Task API.
|
||||
/// backed by Durable Entities), a sub-workflow (dispatched as a sub-orchestration), or a
|
||||
/// regular activity, and invokes the appropriate Durable Task API.
|
||||
/// The serialised string result is returned to the runner for the routing phase.
|
||||
/// </remarks>
|
||||
internal static class DurableExecutorDispatcher
|
||||
{
|
||||
/// <summary>
|
||||
/// Dispatches an executor based on its type (activity or AI agent).
|
||||
/// Dispatches an executor based on its type (activity, AI agent, or sub-workflow).
|
||||
/// </summary>
|
||||
/// <param name="context">The task orchestration context.</param>
|
||||
/// <param name="executorInfo">Information about the executor to dispatch.</param>
|
||||
@@ -48,6 +49,11 @@ internal static class DurableExecutorDispatcher
|
||||
return await ExecuteAgentAsync(context, executorInfo, logger, envelope.Message).ConfigureAwait(true);
|
||||
}
|
||||
|
||||
if (executorInfo.IsSubworkflowExecutor)
|
||||
{
|
||||
return await ExecuteSubWorkflowAsync(context, executorInfo, envelope.Message).ConfigureAwait(true);
|
||||
}
|
||||
|
||||
return await ExecuteActivityAsync(context, executorInfo, envelope.Message, envelope.InputTypeName, sharedState).ConfigureAwait(true);
|
||||
}
|
||||
|
||||
@@ -100,4 +106,76 @@ internal static class DurableExecutorDispatcher
|
||||
|
||||
return response.Text;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Dispatches a sub-workflow executor as a sub-orchestration.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Sub-workflows run as separate orchestration instances, providing independent
|
||||
/// checkpointing, replay, and hierarchical visualization in the DTS dashboard.
|
||||
/// The input is wrapped in <see cref="DurableWorkflowInput{T}"/> to match the
|
||||
/// orchestration's registered input type. The sub-orchestration returns a
|
||||
/// <see cref="DurableWorkflowResult"/> JSON envelope (same as top-level workflows),
|
||||
/// which this method converts to a <see cref="DurableExecutorOutput"/> so the parent
|
||||
/// workflow's result processing picks up both the result and any accumulated events.
|
||||
/// </remarks>
|
||||
private static async Task<string> ExecuteSubWorkflowAsync(
|
||||
TaskOrchestrationContext context,
|
||||
WorkflowExecutorInfo executorInfo,
|
||||
string input)
|
||||
{
|
||||
string orchestrationName = WorkflowNamingHelper.ToOrchestrationFunctionName(executorInfo.SubWorkflow!.Name!);
|
||||
|
||||
DurableWorkflowInput<string> workflowInput = new() { Input = input };
|
||||
|
||||
string? rawOutput = await context.CallSubOrchestratorAsync<string?>(
|
||||
orchestrationName,
|
||||
workflowInput).ConfigureAwait(true);
|
||||
|
||||
return ConvertWorkflowResultToExecutorOutput(rawOutput);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Converts a <see cref="DurableWorkflowResult"/> JSON envelope from a sub-orchestration
|
||||
/// into a <see cref="DurableExecutorOutput"/> JSON string. This bridges the sub-workflow's
|
||||
/// output format to the parent workflow's result processing, preserving both the result
|
||||
/// and any accumulated events from the sub-workflow.
|
||||
/// </summary>
|
||||
private static string ConvertWorkflowResultToExecutorOutput(string? rawOutput)
|
||||
{
|
||||
if (string.IsNullOrEmpty(rawOutput))
|
||||
{
|
||||
return string.Empty;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
DurableWorkflowResult? workflowResult = JsonSerializer.Deserialize(
|
||||
rawOutput,
|
||||
DurableWorkflowJsonContext.Default.DurableWorkflowResult);
|
||||
|
||||
if (workflowResult is null)
|
||||
{
|
||||
return string.Empty;
|
||||
}
|
||||
|
||||
// Propagate the result, events, and sent messages from the sub-workflow.
|
||||
// SentMessages carry the sub-workflow's output for typed routing in the parent,
|
||||
// matching the in-process WorkflowHostExecutor behavior.
|
||||
// Shared state is not included because each workflow instance maintains its own
|
||||
// independent shared state; it is not shared between parent and sub-workflows.
|
||||
DurableExecutorOutput executorOutput = new()
|
||||
{
|
||||
Result = workflowResult.Result,
|
||||
Events = workflowResult.Events ?? [],
|
||||
SentMessages = workflowResult.SentMessages ?? [],
|
||||
};
|
||||
|
||||
return JsonSerializer.Serialize(executorOutput, DurableWorkflowJsonContext.Default.DurableExecutorOutput);
|
||||
}
|
||||
catch (JsonException)
|
||||
{
|
||||
return rawOutput;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -21,4 +21,13 @@ internal sealed class DurableWorkflowResult
|
||||
/// Gets or sets the serialized workflow events emitted during execution.
|
||||
/// </summary>
|
||||
public List<string> Events { get; set; } = [];
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the typed messages to forward to connected executors in the parent workflow.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// When this workflow runs as a sub-orchestration, these messages are propagated to the
|
||||
/// parent workflow and routed to successor executors via the edge map.
|
||||
/// </remarks>
|
||||
public List<TypedPayload> SentMessages { get; set; } = [];
|
||||
}
|
||||
|
||||
@@ -200,10 +200,15 @@ internal sealed class DurableWorkflowRunner
|
||||
// Return wrapper with both result and events so streaming clients can
|
||||
// retrieve events from SerializedOutput after the orchestration completes
|
||||
// (SerializedCustomStatus is cleared by the framework on completion).
|
||||
// SentMessages carries the final result so parent workflows can route it
|
||||
// to connected executors, matching the in-process WorkflowHostExecutor behavior.
|
||||
DurableWorkflowResult workflowResult = new()
|
||||
{
|
||||
Result = finalResult,
|
||||
Events = state.AccumulatedEvents
|
||||
Events = state.AccumulatedEvents,
|
||||
SentMessages = !string.IsNullOrEmpty(finalResult)
|
||||
? [new TypedPayload { Data = finalResult }]
|
||||
: []
|
||||
};
|
||||
|
||||
return JsonSerializer.Serialize(workflowResult, DurableWorkflowJsonContext.Default.DurableWorkflowResult);
|
||||
@@ -600,10 +605,10 @@ internal sealed class DurableWorkflowRunner
|
||||
private static bool HasMeaningfulContent(DurableExecutorOutput output)
|
||||
{
|
||||
return output.Result is not null
|
||||
|| output.SentMessages.Count > 0
|
||||
|| output.Events.Count > 0
|
||||
|| output.StateUpdates.Count > 0
|
||||
|| output.ClearedScopes.Count > 0
|
||||
|| output.SentMessages?.Count > 0
|
||||
|| output.Events?.Count > 0
|
||||
|| output.StateUpdates?.Count > 0
|
||||
|| output.ClearedScopes?.Count > 0
|
||||
|| output.HaltRequested;
|
||||
}
|
||||
}
|
||||
|
||||
+77
@@ -375,6 +375,83 @@ public sealed class WorkflowConsoleAppSamplesValidation(ITestOutputHelper output
|
||||
});
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SubWorkflowsSampleValidationAsync()
|
||||
{
|
||||
using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts();
|
||||
string samplePath = Path.Combine(s_samplesPath, "07_SubWorkflows");
|
||||
|
||||
await this.RunSampleTestAsync(samplePath, async (process, logs) =>
|
||||
{
|
||||
bool inputSent = false;
|
||||
bool foundOrderReceived = false;
|
||||
bool foundValidatePayment = false;
|
||||
bool foundAnalyzePatterns = false;
|
||||
bool foundCalculateRiskScore = false;
|
||||
bool foundChargePayment = false;
|
||||
bool foundSelectCarrier = false;
|
||||
bool foundCreateShipment = false;
|
||||
bool foundOrderCompleted = false;
|
||||
bool foundFraudRiskEvent = false;
|
||||
bool workflowCompleted = false;
|
||||
|
||||
string? line;
|
||||
while ((line = this.ReadLogLine(logs, testTimeoutCts.Token)) != null)
|
||||
{
|
||||
if (!inputSent && line.Contains("Enter an order ID", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
await this.WriteInputAsync(process, "ORD-001", testTimeoutCts.Token);
|
||||
inputSent = true;
|
||||
}
|
||||
|
||||
if (inputSent)
|
||||
{
|
||||
// Main workflow executors
|
||||
foundOrderReceived |= line.Contains("[OrderReceived]", StringComparison.Ordinal);
|
||||
foundOrderCompleted |= line.Contains("[OrderCompleted]", StringComparison.Ordinal);
|
||||
|
||||
// Payment sub-workflow executors
|
||||
foundValidatePayment |= line.Contains("[Payment/ValidatePayment]", StringComparison.Ordinal);
|
||||
foundChargePayment |= line.Contains("[Payment/ChargePayment]", StringComparison.Ordinal);
|
||||
|
||||
// FraudCheck sub-sub-workflow executors (nested inside Payment)
|
||||
foundAnalyzePatterns |= line.Contains("[Payment/FraudCheck/AnalyzePatterns]", StringComparison.Ordinal);
|
||||
foundCalculateRiskScore |= line.Contains("[Payment/FraudCheck/CalculateRiskScore]", StringComparison.Ordinal);
|
||||
|
||||
// Shipping sub-workflow executors
|
||||
foundSelectCarrier |= line.Contains("[Shipping/SelectCarrier]", StringComparison.Ordinal);
|
||||
foundCreateShipment |= line.Contains("[Shipping/CreateShipment]", StringComparison.Ordinal);
|
||||
|
||||
// Custom event from nested sub-workflow
|
||||
foundFraudRiskEvent |= line.Contains("FraudRiskAssessedEvent", StringComparison.Ordinal)
|
||||
|| line.Contains("Risk score", StringComparison.Ordinal);
|
||||
|
||||
if (line.Contains("Order completed", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
workflowCompleted = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
this.AssertNoError(line);
|
||||
}
|
||||
|
||||
Assert.True(inputSent, "Input was not sent to the workflow.");
|
||||
Assert.True(foundOrderReceived, "OrderReceived executor log not found.");
|
||||
Assert.True(foundValidatePayment, "Payment/ValidatePayment executor log not found.");
|
||||
Assert.True(foundAnalyzePatterns, "Payment/FraudCheck/AnalyzePatterns executor log not found.");
|
||||
Assert.True(foundCalculateRiskScore, "Payment/FraudCheck/CalculateRiskScore executor log not found.");
|
||||
Assert.True(foundChargePayment, "Payment/ChargePayment executor log not found.");
|
||||
Assert.True(foundSelectCarrier, "Shipping/SelectCarrier executor log not found.");
|
||||
Assert.True(foundCreateShipment, "Shipping/CreateShipment executor log not found.");
|
||||
Assert.True(foundOrderCompleted, "OrderCompleted executor log not found.");
|
||||
Assert.True(foundFraudRiskEvent, "FraudRiskAssessedEvent from nested sub-workflow not found.");
|
||||
Assert.True(workflowCompleted, "Workflow did not complete successfully.");
|
||||
|
||||
await this.WriteInputAsync(process, "exit", testTimeoutCts.Token);
|
||||
});
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task WorkflowAndAgentsSampleValidationAsync()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user