mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
WIP
This commit is contained in:
Shyju Krishnankutty
@@ -112,14 +112,14 @@
|
||||
<PackageVersion Include="Microsoft.Bot.ObjectModel.PowerFx" Version="1.2025.1106.1" />
|
||||
<PackageVersion Include="Microsoft.PowerFx.Interpreter" Version="1.5.0-build.20251008-1002" />
|
||||
<!-- Durable Task -->
|
||||
<PackageVersion Include="Microsoft.DurableTask.Client" Version="1.18.0" />
|
||||
<PackageVersion Include="Microsoft.DurableTask.Client.AzureManaged" Version="1.18.0" />
|
||||
<PackageVersion Include="Microsoft.DurableTask.Worker" Version="1.18.0" />
|
||||
<PackageVersion Include="Microsoft.DurableTask.Worker.AzureManaged" Version="1.18.0" />
|
||||
<PackageVersion Include="Microsoft.DurableTask.Client" Version="1.19.1" />
|
||||
<PackageVersion Include="Microsoft.DurableTask.Client.AzureManaged" Version="1.19.0" />
|
||||
<PackageVersion Include="Microsoft.DurableTask.Worker" Version="1.19.0" />
|
||||
<PackageVersion Include="Microsoft.DurableTask.Worker.AzureManaged" Version="1.19.0" />
|
||||
<!-- Azure Functions -->
|
||||
<PackageVersion Include="Microsoft.Azure.Functions.Worker" Version="2.50.0" />
|
||||
<PackageVersion Include="Microsoft.Azure.Functions.Worker.ApplicationInsights" Version="2.50.0" />
|
||||
<PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask" Version="1.11.0" />
|
||||
<PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask" Version="1.13.1" />
|
||||
<PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask.AzureManaged" Version="1.0.1" />
|
||||
<PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.Http" Version="3.3.0" />
|
||||
<PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.Http.AspNetCore" Version="2.1.0" />
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
<PropertyGroup>
|
||||
<TargetFrameworks>net10.0</TargetFrameworks>
|
||||
<AzureFunctionsVersion>v4</AzureFunctionsVersion>
|
||||
<OutputType>Exe</OutputType>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<Nullable>enable</Nullable>
|
||||
<!-- The Functions build tools don't like namespaces that start with a number -->
|
||||
<AssemblyName>SingleAgent</AssemblyName>
|
||||
<RootNamespace>SingleAgent</RootNamespace>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<FrameworkReference Include="Microsoft.AspNetCore.App" />
|
||||
</ItemGroup>
|
||||
|
||||
<!-- Azure Functions packages -->
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker" />
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask" />
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask.AzureManaged" />
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.Http" />
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.Http.AspNetCore" />
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker.Sdk" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Azure.AI.OpenAI" />
|
||||
<PackageReference Include="Azure.Identity" />
|
||||
</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.Hosting.AzureFunctions" />
|
||||
<PackageReference Include="Microsoft.Agents.AI.OpenAI" />
|
||||
</ItemGroup>
|
||||
-->
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\..\src\Microsoft.Agents.AI.Hosting.AzureFunctions\Microsoft.Agents.AI.Hosting.AzureFunctions.csproj" />
|
||||
<ProjectReference Include="..\..\..\src\Microsoft.Agents.AI.OpenAI\Microsoft.Agents.AI.OpenAI.csproj" />
|
||||
<ProjectReference Include="..\..\..\src\Microsoft.Agents.AI.Workflows\Microsoft.Agents.AI.Workflows.csproj" />
|
||||
</ItemGroup>
|
||||
</Project>
|
||||
@@ -0,0 +1,38 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
|
||||
namespace SingleAgent;
|
||||
|
||||
internal sealed class ConcurrentStartExecutor() : Executor<string, string>("ConcurrentStartExecutor")
|
||||
{
|
||||
public override ValueTask<string> 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<string[], string>("ConcurrentAggregationExecutor")
|
||||
{
|
||||
/// <summary>
|
||||
/// Handles incoming messages from the agents and aggregates their responses.
|
||||
/// </summary>
|
||||
/// <param name="message">The messages from the parallel agents.</param>
|
||||
/// <param name="context">Workflow context for accessing workflow services and adding events.</param>
|
||||
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests.
|
||||
/// The default is <see cref="CancellationToken.None"/>.</param>
|
||||
/// <returns>A task representing the asynchronous operation.</returns>
|
||||
public override ValueTask<string> 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}");
|
||||
}
|
||||
}
|
||||
@@ -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();
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
@@ -0,0 +1,27 @@
|
||||
//// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
//using Microsoft.Agents.AI.Workflows;
|
||||
|
||||
//namespace SingleAgent;
|
||||
|
||||
///// <summary>
|
||||
///// Routes survey responses to appropriate teams based on rating and category.
|
||||
///// </summary>
|
||||
//public sealed class ResponseRouterExecutor() : Executor<string, string>("ResponseRouterExecutor")
|
||||
//{
|
||||
// public override ValueTask<string> 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");
|
||||
// }
|
||||
// }
|
||||
//}
|
||||
@@ -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;
|
||||
|
||||
///// <summary>
|
||||
///// This executor parses survey responses and produces structured output.
|
||||
///// Example input: "Rating: 8. The app is good but checkout process is confusing."
|
||||
///// </summary>
|
||||
//[System.Diagnostics.CodeAnalysis.SuppressMessage("Performance", "CA1812:Avoid uninstantiated internal classes", Justification = "Instantiated by workflow framework")]
|
||||
//internal sealed partial class SurveyResponseParserExecutor() : Executor<string, string>("SurveyResponseParserExecutor")
|
||||
//{
|
||||
// private static readonly JsonSerializerOptions s_jsonOptions = new()
|
||||
// {
|
||||
// WriteIndented = true
|
||||
// };
|
||||
|
||||
// [GeneratedRegex(@"Rating:\s*(\d+)", RegexOptions.IgnoreCase)]
|
||||
// private static partial Regex RatingRegex();
|
||||
|
||||
// public override ValueTask<string> 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;
|
||||
// }
|
||||
//}
|
||||
@@ -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?
|
||||
@@ -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"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+44
@@ -0,0 +1,44 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
<PropertyGroup>
|
||||
<TargetFrameworks>net10.0</TargetFrameworks>
|
||||
<AzureFunctionsVersion>v4</AzureFunctionsVersion>
|
||||
<OutputType>Exe</OutputType>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<Nullable>enable</Nullable>
|
||||
<!-- The Functions build tools don't like namespaces that start with a number -->
|
||||
<AssemblyName>SingleAgent</AssemblyName>
|
||||
<RootNamespace>SingleAgent</RootNamespace>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<FrameworkReference Include="Microsoft.AspNetCore.App" />
|
||||
</ItemGroup>
|
||||
|
||||
<!-- Azure Functions packages -->
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker" />
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask" />
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask.AzureManaged" />
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.Http" />
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.Http.AspNetCore" />
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker.Sdk" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Azure.AI.OpenAI" />
|
||||
<PackageReference Include="Azure.Identity" />
|
||||
</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.Hosting.AzureFunctions" />
|
||||
<PackageReference Include="Microsoft.Agents.AI.OpenAI" />
|
||||
</ItemGroup>
|
||||
-->
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\..\src\Microsoft.Agents.AI.Hosting.AzureFunctions\Microsoft.Agents.AI.Hosting.AzureFunctions.csproj" />
|
||||
<ProjectReference Include="..\..\..\src\Microsoft.Agents.AI.OpenAI\Microsoft.Agents.AI.OpenAI.csproj" />
|
||||
<ProjectReference Include="..\..\..\src\Microsoft.Agents.AI.Workflows\Microsoft.Agents.AI.Workflows.csproj" />
|
||||
</ItemGroup>
|
||||
</Project>
|
||||
+215
@@ -0,0 +1,215 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
|
||||
namespace SingleAgent;
|
||||
|
||||
/// <summary>
|
||||
/// Represents the details of a customer order.
|
||||
/// </summary>
|
||||
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<OrderItem> Items { get; set; } = [];
|
||||
|
||||
public decimal TotalAmount { get; set; }
|
||||
|
||||
public OrderStatus Status { get; set; }
|
||||
|
||||
public DateTime OrderDate { get; set; }
|
||||
|
||||
public DateTime? EstimatedDelivery { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Represents an item in an order.
|
||||
/// </summary>
|
||||
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;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Represents the status of an order.
|
||||
/// </summary>
|
||||
public enum OrderStatus
|
||||
{
|
||||
Pending,
|
||||
Processing,
|
||||
Shipped,
|
||||
Delivered,
|
||||
Cancelled
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Executor that looks up order details by order ID.
|
||||
/// Input: int (orderId)
|
||||
/// Output: OrderDetails
|
||||
/// </summary>
|
||||
internal sealed class OrderLookupExecutor() : Executor<int, OrderDetails>("OrderLookupExecutor")
|
||||
{
|
||||
// Simulated order database
|
||||
private static readonly Dictionary<int, OrderDetails> 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<OrderDetails> 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
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Executor that generates a human-readable summary from order details.
|
||||
/// Input: OrderDetails
|
||||
/// Output: string
|
||||
/// </summary>
|
||||
internal sealed class OrderSummaryExecutor() : Executor<OrderDetails, string>("OrderSummaryExecutor")
|
||||
{
|
||||
public override ValueTask<string> 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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <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)
|
||||
/// </summary>
|
||||
internal sealed class OrderIdParserExecutor() : Executor<string, int>("OrderIdParserExecutor")
|
||||
{
|
||||
public override ValueTask<int> 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);
|
||||
}
|
||||
}
|
||||
@@ -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();
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
@@ -0,0 +1,27 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
|
||||
namespace SingleAgent;
|
||||
|
||||
/// <summary>
|
||||
/// Routes survey responses to appropriate teams based on rating and category.
|
||||
/// </summary>
|
||||
public sealed class ResponseRouterExecutor() : Executor<string, string>("ResponseRouterExecutor")
|
||||
{
|
||||
public override ValueTask<string> 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");
|
||||
}
|
||||
}
|
||||
}
|
||||
+82
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// This executor parses survey responses and produces structured output.
|
||||
/// Example input: "Rating: 8. The app is good but checkout process is confusing."
|
||||
/// </summary>
|
||||
[System.Diagnostics.CodeAnalysis.SuppressMessage("Performance", "CA1812:Avoid uninstantiated internal classes", Justification = "Instantiated by workflow framework")]
|
||||
internal sealed partial class SurveyResponseParserExecutor() : Executor<string, string>("SurveyResponseParserExecutor")
|
||||
{
|
||||
private static readonly JsonSerializerOptions s_jsonOptions = new()
|
||||
{
|
||||
WriteIndented = true
|
||||
};
|
||||
|
||||
[GeneratedRegex(@"Rating:\s*(\d+)", RegexOptions.IgnoreCase)]
|
||||
private static partial Regex RatingRegex();
|
||||
|
||||
public override ValueTask<string> 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;
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
@@ -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"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -10,6 +10,7 @@ public sealed class DurableAgentsOptions
|
||||
// Agent names are case-insensitive
|
||||
private readonly Dictionary<string, Func<IServiceProvider, AIAgent>> _agentFactories = new(StringComparer.OrdinalIgnoreCase);
|
||||
private readonly Dictionary<string, TimeSpan?> _agentTimeToLive = new(StringComparer.OrdinalIgnoreCase);
|
||||
private readonly HashSet<string> _workflowOnlyAgents = new(StringComparer.OrdinalIgnoreCase);
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="DurableAgentsOptions"/> class.
|
||||
@@ -104,6 +105,22 @@ public sealed class DurableAgentsOptions
|
||||
/// Thrown when <paramref name="agent.Name"/> is null or whitespace or when an agent with the same name has already been registered.
|
||||
/// </exception>
|
||||
public DurableAgentsOptions AddAIAgent(AIAgent agent, TimeSpan? timeToLive = null)
|
||||
{
|
||||
return this.AddAIAgent(agent, workflowOnly: false, timeToLive);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Adds an AI agent to the options with workflow-only configuration.
|
||||
/// </summary>
|
||||
/// <param name="agent">The agent to add.</param>
|
||||
/// <param name="workflowOnly">If true, the agent is only accessible within workflows and won't have HTTP triggers.</param>
|
||||
/// <param name="timeToLive">Optional time-to-live for this agent's entities. If not specified, uses <see cref="DefaultTimeToLive"/>.</param>
|
||||
/// <returns>The options instance.</returns>
|
||||
/// <exception cref="ArgumentNullException">Thrown when <paramref name="agent"/> is null.</exception>
|
||||
/// <exception cref="ArgumentException">
|
||||
/// Thrown when <paramref name="agent.Name"/> is null or whitespace or when an agent with the same name has already been registered.
|
||||
/// </exception>
|
||||
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;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Determines whether an agent is configured as workflow-only (no HTTP triggers).
|
||||
/// </summary>
|
||||
/// <param name="agentName">The name of the agent.</param>
|
||||
/// <returns>True if the agent is workflow-only; otherwise, false.</returns>
|
||||
internal bool IsWorkflowOnly(string agentName)
|
||||
{
|
||||
return this._workflowOnlyAgents.Contains(agentName);
|
||||
}
|
||||
}
|
||||
|
||||
+2
-4
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Provides configuration options for durable features in Azure Functions.
|
||||
/// Provides configuration options for durable agents and workflows.
|
||||
/// </summary>
|
||||
public sealed class DurableOptions
|
||||
{
|
||||
@@ -10,17 +10,31 @@ namespace Microsoft.Agents.AI.DurableTask;
|
||||
public sealed class DurableWorkflowOptions
|
||||
{
|
||||
private readonly Dictionary<string, Workflow> _workflows = new(StringComparer.OrdinalIgnoreCase);
|
||||
private readonly object? _parentOptions;
|
||||
private readonly DurableOptions? _parentOptions;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="DurableWorkflowOptions"/> class.
|
||||
/// </summary>
|
||||
/// <param name="parentOptions">Optional parent options container for accessing related configuration.</param>
|
||||
public DurableWorkflowOptions(object? parentOptions = null)
|
||||
internal DurableWorkflowOptions(DurableOptions? parentOptions = null)
|
||||
{
|
||||
this._parentOptions = parentOptions;
|
||||
this.Executors = new ExecutorRegistry();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets the collection of workflows available in the current context, keyed by their unique names.
|
||||
/// </summary>
|
||||
/// <remarks>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.</remarks>
|
||||
public IReadOnlyDictionary<string, Workflow> Workflows => this._workflows;
|
||||
|
||||
/// <summary>
|
||||
/// Gets the executor registry.
|
||||
/// </summary>
|
||||
internal ExecutorRegistry Executors { get; }
|
||||
|
||||
/// <summary>
|
||||
/// Adds a workflow to the collection for processing or execution.
|
||||
/// </summary>
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets the collection of workflows available in the current context, keyed by their unique names.
|
||||
/// </summary>
|
||||
/// <remarks>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.</remarks>
|
||||
public IReadOnlyDictionary<string, Workflow> 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<string, Workflows.Checkpointing.ExecutorInfo> executor in workflow.ReflectExecutors())
|
||||
{
|
||||
// Extract the executor name (without GUID suffix)
|
||||
string executorName = executor.Key.Split('_')[0];
|
||||
registry.Register(executorName, executor.Key, workflow);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
|
||||
namespace Microsoft.Agents.AI.DurableTask;
|
||||
|
||||
/// <summary>
|
||||
/// Provides a registry for storing and retrieving executor bindings independently from workflows.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 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.
|
||||
/// </remarks>
|
||||
internal sealed class ExecutorRegistry
|
||||
{
|
||||
private readonly Dictionary<string, ExecutorRegistration> _executors = new(StringComparer.OrdinalIgnoreCase);
|
||||
|
||||
/// <summary>
|
||||
/// Registers an executor binding from a workflow.
|
||||
/// </summary>
|
||||
/// <param name="executorName">The executor name (without GUID suffix).</param>
|
||||
/// <param name="executorId">The full executor ID (may include GUID suffix).</param>
|
||||
/// <param name="workflow">The workflow containing the executor.</param>
|
||||
/// <remarks>
|
||||
/// 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.
|
||||
/// </remarks>
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Attempts to get an executor registration by name.
|
||||
/// </summary>
|
||||
/// <param name="executorName">The executor name to look up.</param>
|
||||
/// <param name="registration">When this method returns, contains the registration if found; otherwise, null.</param>
|
||||
/// <returns><c>true</c> if the executor was found; otherwise, <c>false</c>.</returns>
|
||||
public bool TryGetExecutor(string executorName, out ExecutorRegistration? registration)
|
||||
{
|
||||
return this._executors.TryGetValue(executorName, out registration);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets the number of registered executors.
|
||||
/// </summary>
|
||||
public int Count => this._executors.Count;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Represents a registered executor with its associated workflow.
|
||||
/// </summary>
|
||||
/// <param name="ExecutorId">The full executor ID (may include GUID suffix).</param>
|
||||
/// <param name="Workflow">The workflow containing the executor.</param>
|
||||
internal sealed record ExecutorRegistration(string ExecutorId, Workflow Workflow)
|
||||
{
|
||||
/// <summary>
|
||||
/// Creates an instance of the executor.
|
||||
/// </summary>
|
||||
/// <param name="runId">A unique identifier for the run context.</param>
|
||||
/// <param name="cancellationToken">The cancellation token.</param>
|
||||
/// <returns>The created executor instance.</returns>
|
||||
public ValueTask<Executor> CreateExecutorInstanceAsync(string runId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
return this.Workflow.CreateExecutorInstanceAsync(this.ExecutorId, runId, cancellationToken);
|
||||
}
|
||||
}
|
||||
+24
-1
@@ -113,6 +113,29 @@ public static class DurableAgentsOptionsExtensions
|
||||
Func<IServiceProvider, AIAgent> factory,
|
||||
bool enableHttpTrigger,
|
||||
bool enableMcpToolTrigger)
|
||||
{
|
||||
return AddAIAgentFactory(options, name, factory, enableHttpTrigger, enableMcpToolTrigger, timeToLive: null);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Registers an AI agent factory with the specified name, trigger options, and time-to-live configuration.
|
||||
/// </summary>
|
||||
/// <remarks>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.</remarks>
|
||||
/// <param name="options">The options object to which the AI agent factory will be added. Cannot be null.</param>
|
||||
/// <param name="name">The unique name used to identify the AI agent factory. Cannot be null.</param>
|
||||
/// <param name="factory">A delegate that creates an instance of the AI agent using the provided service provider. Cannot be null.</param>
|
||||
/// <param name="enableHttpTrigger">true to enable the HTTP trigger for the agent; otherwise, false.</param>
|
||||
/// <param name="enableMcpToolTrigger">true to enable the MCP tool trigger for the agent; otherwise, false.</param>
|
||||
/// <param name="timeToLive">Optional time-to-live for this agent's entities.</param>
|
||||
/// <returns>The same DurableAgentsOptions instance, allowing for method chaining.</returns>
|
||||
public static DurableAgentsOptions AddAIAgentFactory(
|
||||
this DurableAgentsOptions options,
|
||||
string name,
|
||||
Func<IServiceProvider, AIAgent> 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;
|
||||
}
|
||||
|
||||
@@ -26,78 +26,90 @@ public static class DurableOptionsExtensions
|
||||
/// <returns>The Functions application builder for method chaining.</returns>
|
||||
/// <remarks>
|
||||
/// 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.
|
||||
/// </remarks>
|
||||
public static FunctionsApplicationBuilder ConfigureDurableOptions(
|
||||
this FunctionsApplicationBuilder builder,
|
||||
Action<DurableOptions> 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<string, Func<IServiceProvider, AIAgent>> 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<IFunctionsAgentOptionsProvider>(_ =>
|
||||
new DefaultFunctionsAgentOptionsProvider(DurableAgentsOptionsExtensions.GetAgentOptionsSnapshot()));
|
||||
|
||||
builder.Services.AddSingleton<IFunctionMetadataTransformer, DurableAgentFunctionMetadataTransformer>();
|
||||
|
||||
// Handling of built-in function execution for Agent HTTP, MCP tool, or Entity invocations.
|
||||
builder.UseWhen<BuiltInFunctionExecutionMiddleware>(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<BuiltInFunctionExecutor>();
|
||||
|
||||
builder.Services.AddSingleton<DurableWorkflowRunner>();
|
||||
|
||||
builder.ConfigureDurableWorker().AddTasks(t => t.AddOrchestratorFunc<DuableWorkflowRunRequest, List<string>>(
|
||||
"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<DurableWorkflowRunner>();
|
||||
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<IFunctionMetadataTransformer, DurableWorkflowFunctionMetadataTransformer>();
|
||||
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<BuiltInFunctionExecutor>();
|
||||
builder.Services.AddSingleton<DurableWorkflowRunner>();
|
||||
builder.Services.AddSingleton<IFunctionMetadataTransformer, DurableAgentFunctionMetadataTransformer>();
|
||||
builder.Services.AddSingleton<IFunctionMetadataTransformer, DurableWorkflowFunctionMetadataTransformer>();
|
||||
|
||||
builder.Services.TryAddSingleton<IFunctionsAgentOptionsProvider>(_ =>
|
||||
new DefaultFunctionsAgentOptionsProvider(DurableAgentsOptionsExtensions.GetAgentOptionsSnapshot()));
|
||||
}
|
||||
|
||||
private static void ConfigureAgents(FunctionsApplicationBuilder builder, DurableOptions options)
|
||||
{
|
||||
builder.Services.ConfigureDurableAgents(agentOpts =>
|
||||
{
|
||||
foreach (KeyValuePair<string, Func<IServiceProvider, AIAgent>> 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<BuiltInFunctionExecutionMiddleware>(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<DuableWorkflowRunRequest, List<string>>(
|
||||
"WorkflowRunnerOrchestration",
|
||||
async (orchestrationContext, request) =>
|
||||
{
|
||||
FunctionContext functionContext = orchestrationContext.GetFunctionContext()
|
||||
?? throw new InvalidOperationException("FunctionContext is not available in the orchestration context.");
|
||||
|
||||
DurableWorkflowRunner runner = functionContext.InstanceServices.GetRequiredService<DurableWorkflowRunner>();
|
||||
ILogger logger = orchestrationContext.CreateReplaySafeLogger("WorkflowRunnerOrchestration");
|
||||
|
||||
return await runner.RunWorkflowOrchestrationAsync(orchestrationContext, request, logger).ConfigureAwait(true);
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
||||
+20
-18
@@ -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,
|
||||
// };
|
||||
//}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Executes workflow orchestrations and activity functions for durable workflows.
|
||||
/// </summary>
|
||||
internal sealed class DurableWorkflowRunner
|
||||
{
|
||||
private readonly DurableWorkflowOptions _options;
|
||||
private readonly ILogger<DurableWorkflowRunner> _logger;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="DurableWorkflowRunner"/> class.
|
||||
/// </summary>
|
||||
/// <param name="logger">The logger instance.</param>
|
||||
/// <param name="durableOptions">The durable options containing workflow configurations.</param>
|
||||
public DurableWorkflowRunner(ILogger<DurableWorkflowRunner> logger, DurableOptions durableOptions)
|
||||
{
|
||||
this._logger = logger;
|
||||
ArgumentNullException.ThrowIfNull(logger);
|
||||
ArgumentNullException.ThrowIfNull(durableOptions);
|
||||
|
||||
this._logger = logger;
|
||||
this._options = durableOptions.Workflows;
|
||||
}
|
||||
|
||||
internal async Task<List<string>> RunWorkflowOrchestrationAsync(TaskOrchestrationContext taskOrchestrationContext, DuableWorkflowRunRequest input, ILogger logger)
|
||||
/// <summary>
|
||||
/// Runs a workflow orchestration.
|
||||
/// </summary>
|
||||
/// <param name="context">The task orchestration context.</param>
|
||||
/// <param name="request">The workflow run request containing workflow name and input.</param>
|
||||
/// <param name="logger">The replay-safe logger for orchestration logging.</param>
|
||||
/// <returns>A list containing the workflow execution result.</returns>
|
||||
internal async Task<List<string>> 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<string> RunExecutorsInWorkFlowAsync(
|
||||
TaskOrchestrationContext taskOrchestrationContext,
|
||||
Workflow workflow,
|
||||
string initialInput,
|
||||
ILogger logger)
|
||||
{
|
||||
List<string> 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<string>(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<string> ExecuteActivityAsync(string activityFunctionName, string input, FunctionContext functionContext)
|
||||
/// <summary>
|
||||
/// Executes an activity function for a workflow executor.
|
||||
/// </summary>
|
||||
/// <param name="activityFunctionName">The name of the activity function to execute.</param>
|
||||
/// <param name="input">The input string for the executor.</param>
|
||||
/// <param name="functionContext">The Azure Functions context.</param>
|
||||
/// <returns>The serialized result of the executor.</returns>
|
||||
internal async Task<string> 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<string> ExecuteWorkflowLevelsAsync(
|
||||
TaskOrchestrationContext context,
|
||||
Workflow workflow,
|
||||
string initialInput,
|
||||
ILogger logger)
|
||||
{
|
||||
WorkflowExecutionPlan plan = WorkflowHelper.GetExecutionPlan(workflow);
|
||||
Dictionary<string, string> 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<Task<(string Id, string Result)>> 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<string> 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<string>(triggerName, input).ConfigureAwait(true);
|
||||
}
|
||||
|
||||
// Get the executor info
|
||||
Dictionary<string, ExecutorInfo> 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<string, ExecutorInfo> executorPair = executorInfos.FirstOrDefault(e =>
|
||||
e.Key.StartsWith(executorName + "_", StringComparison.Ordinal) ||
|
||||
string.Equals(e.Key, executorName, StringComparison.Ordinal));
|
||||
private async Task<string> 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<string, string> results,
|
||||
WorkflowExecutionPlan plan)
|
||||
{
|
||||
List<string> 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<string> 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<string, string> results)
|
||||
{
|
||||
WorkflowExecutionLevel lastLevel = plan.Levels[^1];
|
||||
|
||||
if (lastLevel.Executors.Count == 1)
|
||||
{
|
||||
return results[lastLevel.Executors[0].ExecutorId];
|
||||
}
|
||||
|
||||
List<string> 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<string> 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<string>(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}'.");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,6 +12,46 @@ namespace Microsoft.Agents.AI.Hosting.AzureFunctions;
|
||||
/// <param name="IsAgenticExecutor">Indicates whether this executor is an agentic executor.</param>
|
||||
internal sealed record WorkflowExecutorInfo(string ExecutorId, bool IsAgenticExecutor);
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <param name="Level">The level number (0-based, starting from the root executor).</param>
|
||||
/// <param name="Executors">The executors that can run in parallel at this level.</param>
|
||||
/// <param name="IsFanIn">Indicates if this level is a Fan-In point (has executors with multiple predecessors).</param>
|
||||
internal sealed record WorkflowExecutionLevel(int Level, List<WorkflowExecutorInfo> Executors, bool IsFanIn);
|
||||
|
||||
/// <summary>
|
||||
/// Represents the complete execution plan for a workflow, including parallel execution levels.
|
||||
/// </summary>
|
||||
internal sealed class WorkflowExecutionPlan
|
||||
{
|
||||
/// <summary>
|
||||
/// The execution levels in order. Each level contains executors that can run in parallel.
|
||||
/// </summary>
|
||||
public List<WorkflowExecutionLevel> Levels { get; } = [];
|
||||
|
||||
/// <summary>
|
||||
/// Maps each executor ID to its predecessors (for Fan-In result aggregation).
|
||||
/// </summary>
|
||||
public Dictionary<string, List<string>> Predecessors { get; } = [];
|
||||
|
||||
/// <summary>
|
||||
/// Maps each executor ID to its successors (for Fan-Out result distribution).
|
||||
/// </summary>
|
||||
public Dictionary<string, List<string>> Successors { get; } = [];
|
||||
|
||||
/// <summary>
|
||||
/// Gets whether this workflow has any parallel execution opportunities.
|
||||
/// </summary>
|
||||
public bool HasParallelism => this.Levels.Any(l => l.Executors.Count > 1);
|
||||
|
||||
/// <summary>
|
||||
/// Gets whether this workflow has any Fan-In points.
|
||||
/// </summary>
|
||||
public bool HasFanIn => this.Levels.Any(l => l.IsFanIn);
|
||||
}
|
||||
|
||||
internal static class WorkflowHelper
|
||||
{
|
||||
/// <summary>
|
||||
@@ -20,20 +60,45 @@ internal static class WorkflowHelper
|
||||
/// <param name="workflow">The workflow instance to analyze.</param>
|
||||
/// <returns>A list of executor information in topological order (execution order).</returns>
|
||||
public static List<WorkflowExecutorInfo> GetExecutorsFromWorkflowInOrder(Workflow workflow)
|
||||
{
|
||||
WorkflowExecutionPlan plan = GetExecutionPlan(workflow);
|
||||
|
||||
// Flatten the levels into a single list for backward compatibility
|
||||
List<WorkflowExecutorInfo> result = [];
|
||||
foreach (WorkflowExecutionLevel level in plan.Levels)
|
||||
{
|
||||
result.AddRange(level.Executors);
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <param name="workflow">The workflow instance to analyze.</param>
|
||||
/// <returns>An execution plan with parallel execution levels.</returns>
|
||||
public static WorkflowExecutionPlan GetExecutionPlan(Workflow workflow)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(workflow);
|
||||
|
||||
Dictionary<string, ExecutorInfo> executors = workflow.ReflectExecutors();
|
||||
Dictionary<string, HashSet<EdgeInfo>> edges = workflow.ReflectEdges();
|
||||
|
||||
// Build adjacency list and in-degree map
|
||||
Dictionary<string, List<string>> adjacencyList = new();
|
||||
Dictionary<string, int> inDegree = new();
|
||||
WorkflowExecutionPlan plan = new();
|
||||
|
||||
// Initialize all executors with in-degree 0
|
||||
// Build adjacency lists (successors and predecessors)
|
||||
Dictionary<string, List<string>> successors = [];
|
||||
Dictionary<string, List<string>> predecessors = [];
|
||||
Dictionary<string, int> inDegree = [];
|
||||
|
||||
// Initialize all executors
|
||||
foreach (string executorId in executors.Keys)
|
||||
{
|
||||
adjacencyList[executorId] = new List<string>();
|
||||
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<string> orderedExecutorIds = new();
|
||||
Queue<string> 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<string, int> 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<string> processed = [];
|
||||
Dictionary<string, int> 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<string> currentLevelIds = [];
|
||||
|
||||
// For each neighbor of the current executor
|
||||
foreach (string neighbor in adjacencyList[current])
|
||||
foreach (KeyValuePair<string, int> 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<WorkflowExecutorInfo> 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<WorkflowExecutorInfo> 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;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
Reference in New Issue
Block a user