mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
WIP
This commit is contained in:
@@ -0,0 +1,37 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
|
||||
namespace Microsoft.Agents.AI.DurableTask;
|
||||
|
||||
/// <summary>
|
||||
/// Provides configuration options for managing durable workflows within an application.
|
||||
/// </summary>
|
||||
public sealed class DurableWorkflowOptions
|
||||
{
|
||||
private readonly Dictionary<string, Workflow> _workflows = new(StringComparer.OrdinalIgnoreCase);
|
||||
|
||||
/// <summary>
|
||||
/// Adds a workflow to the collection for processing or execution.
|
||||
/// </summary>
|
||||
/// <param name="workflow">The workflow instance to add. Cannot be null.</param>
|
||||
public void AddWorkflow(Workflow workflow)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(workflow);
|
||||
|
||||
if (string.IsNullOrEmpty(workflow.Name))
|
||||
{
|
||||
throw new ArgumentException("Workflow must have a valid Name property.", nameof(workflow));
|
||||
}
|
||||
|
||||
this._workflows[workflow.Name] = workflow;
|
||||
}
|
||||
|
||||
/// <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;
|
||||
}
|
||||
@@ -24,6 +24,7 @@
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\Microsoft.Agents.AI.Workflows\Microsoft.Agents.AI.Workflows.csproj" />
|
||||
<ProjectReference Include="..\Microsoft.Agents.AI\Microsoft.Agents.AI.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
|
||||
@@ -5,6 +5,7 @@ using Microsoft.Azure.Functions.Worker.Context.Features;
|
||||
using Microsoft.Azure.Functions.Worker.Extensions.Mcp;
|
||||
using Microsoft.Azure.Functions.Worker.Http;
|
||||
using Microsoft.Azure.Functions.Worker.Invocation;
|
||||
using Microsoft.DurableTask;
|
||||
using Microsoft.DurableTask.Client;
|
||||
|
||||
namespace Microsoft.Agents.AI.Hosting.AzureFunctions;
|
||||
@@ -25,6 +26,20 @@ internal sealed class BuiltInFunctionExecutor : IFunctionExecutor
|
||||
IFunctionInputBindingFeature? functionInputBindingFeature = context.Features.Get<IFunctionInputBindingFeature>() ??
|
||||
throw new InvalidOperationException("Function input binding feature is not available on the current context.");
|
||||
|
||||
if (context.FunctionDefinition.EntryPoint == BuiltInFunctions.RunWorkflowOrechstrtationFunctionEntryPoint)
|
||||
{
|
||||
var triggerBinding = context.FunctionDefinition.InputBindings.Values.FirstOrDefault(b => b.Type == "orchestrationTrigger");
|
||||
var taskOrechstrationContextBinding = context.BindInputAsync<TaskOrchestrationContext>(triggerBinding!);
|
||||
|
||||
if (taskOrechstrationContextBinding.IsCompletedSuccessfully)
|
||||
{
|
||||
var t = taskOrechstrationContextBinding.Result.Value;
|
||||
context.GetInvocationResult().Value = await BuiltInFunctions.RunWorkflowOrchestratorAsync(t!);
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
FunctionInputBindingResult? inputBindingResults = await functionInputBindingFeature.BindFunctionInputAsync(context);
|
||||
if (inputBindingResults is not { Values: { } values })
|
||||
{
|
||||
@@ -35,6 +50,7 @@ internal sealed class BuiltInFunctionExecutor : IFunctionExecutor
|
||||
string? encodedEntityRequest = null;
|
||||
DurableTaskClient? durableTaskClient = null;
|
||||
ToolInvocationContext? mcpToolInvocationContext = null;
|
||||
//string? encodedTaskOrchestrationContext = null;
|
||||
|
||||
foreach (var binding in values)
|
||||
{
|
||||
@@ -52,10 +68,13 @@ internal sealed class BuiltInFunctionExecutor : IFunctionExecutor
|
||||
case ToolInvocationContext toolContext:
|
||||
mcpToolInvocationContext = toolContext;
|
||||
break;
|
||||
//case string orchestrationContext:
|
||||
// encodedTaskOrchestrationContext = orchestrationContext;
|
||||
// break;
|
||||
}
|
||||
}
|
||||
|
||||
if (durableTaskClient is null)
|
||||
if (durableTaskClient is null && context.FunctionDefinition.EntryPoint != BuiltInFunctions.RunWorkflowOrechstrtationFunctionEntryPoint)
|
||||
{
|
||||
// This is not expected to happen since all built-in functions are
|
||||
// expected to have a Durable Task client binding.
|
||||
@@ -71,7 +90,7 @@ internal sealed class BuiltInFunctionExecutor : IFunctionExecutor
|
||||
|
||||
context.GetInvocationResult().Value = await BuiltInFunctions.RunAgentHttpAsync(
|
||||
httpRequestData,
|
||||
durableTaskClient,
|
||||
durableTaskClient!,
|
||||
context);
|
||||
return;
|
||||
}
|
||||
@@ -84,7 +103,7 @@ internal sealed class BuiltInFunctionExecutor : IFunctionExecutor
|
||||
}
|
||||
|
||||
context.GetInvocationResult().Value = await BuiltInFunctions.InvokeAgentAsync(
|
||||
durableTaskClient,
|
||||
durableTaskClient!,
|
||||
encodedEntityRequest,
|
||||
context);
|
||||
return;
|
||||
@@ -98,7 +117,40 @@ internal sealed class BuiltInFunctionExecutor : IFunctionExecutor
|
||||
}
|
||||
|
||||
context.GetInvocationResult().Value =
|
||||
await BuiltInFunctions.RunMcpToolAsync(mcpToolInvocationContext, durableTaskClient, context);
|
||||
await BuiltInFunctions.RunMcpToolAsync(mcpToolInvocationContext, durableTaskClient!, context);
|
||||
return;
|
||||
}
|
||||
|
||||
if (context.FunctionDefinition.EntryPoint == BuiltInFunctions.RunWorkflowOrechstrtationHttpFunctionEntryPoint)
|
||||
{
|
||||
//if (httpRequestData == null)
|
||||
//{
|
||||
// throw new InvalidOperationException($"HTTP request data binding is missing for the invocation {context.InvocationId}.");
|
||||
//}
|
||||
|
||||
if (httpRequestData == null)
|
||||
{
|
||||
throw new InvalidOperationException($"HTTP request data binding is missing for the invocation {context.InvocationId}.");
|
||||
}
|
||||
|
||||
context.GetInvocationResult().Value = await BuiltInFunctions.RunWorkflowOrechstrtationHttpTriggerAsync(
|
||||
httpRequestData,
|
||||
durableTaskClient!,
|
||||
context);
|
||||
return;
|
||||
}
|
||||
if (context.FunctionDefinition.EntryPoint == BuiltInFunctions.RunWorkflowOrechstrtationFunctionEntryPoint)
|
||||
{
|
||||
var triggerBinding = context.FunctionDefinition.InputBindings.Values.FirstOrDefault(b => b.Type == "orchestrationTrigger");
|
||||
var taskOrechstrationContextBinding = context.BindInputAsync<TaskOrchestrationContext>(triggerBinding!);
|
||||
|
||||
if (taskOrechstrationContextBinding.IsCompletedSuccessfully)
|
||||
{
|
||||
var t = taskOrechstrationContextBinding.Result.Value;
|
||||
context.GetInvocationResult().Value = await BuiltInFunctions.RunWorkflowOrchestratorAsync(t!);
|
||||
}
|
||||
|
||||
//context.GetInvocationResult().Value = await BuiltInFunctions.RunWorkflowOrchestratorAsync(null);
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@ using Microsoft.Agents.AI.DurableTask;
|
||||
using Microsoft.Azure.Functions.Worker;
|
||||
using Microsoft.Azure.Functions.Worker.Extensions.Mcp;
|
||||
using Microsoft.Azure.Functions.Worker.Http;
|
||||
using Microsoft.DurableTask;
|
||||
using Microsoft.DurableTask.Client;
|
||||
using Microsoft.DurableTask.Worker.Grpc;
|
||||
using Microsoft.Extensions.AI;
|
||||
@@ -20,8 +21,47 @@ internal static class BuiltInFunctions
|
||||
|
||||
internal static readonly string RunAgentHttpFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(RunAgentHttpAsync)}";
|
||||
internal static readonly string RunAgentEntityFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(InvokeAgentAsync)}";
|
||||
internal static readonly string RunWorkflowOrechstrtationHttpFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(RunWorkflowOrechstrtationHttpTriggerAsync)}";
|
||||
internal static readonly string RunWorkflowOrechstrtationFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(RunWorkflowOrchestratorAsync)}";
|
||||
internal static readonly string InvokeWorkflowActivityFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(InvokeWorkflowActivityAsync)}";
|
||||
internal static readonly string RunAgentMcpToolFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(RunMcpToolAsync)}";
|
||||
|
||||
#pragma warning disable IL3000 // Avoid accessing Assembly file path when publishing as a single file - Azure Functions does not use single-file publishing
|
||||
internal static readonly string ScriptFile = Path.GetFileName(typeof(BuiltInFunctions).Assembly.Location);
|
||||
#pragma warning restore IL3000
|
||||
|
||||
// Exposed as an activity trigger for workflow executors
|
||||
public static Task<string> InvokeWorkflowActivityAsync(
|
||||
[ActivityTrigger] string input,
|
||||
FunctionContext functionContext)
|
||||
{
|
||||
return Task.FromResult($"Hello from activity with input: {input}");
|
||||
}
|
||||
|
||||
//[Function("my-Orchestration")]
|
||||
//public static async Task<List<string>> RunOrchestrator1Async(
|
||||
//[OrchestrationTrigger] TaskOrchestrationContext context)
|
||||
//{
|
||||
// ILogger logger = context.CreateReplaySafeLogger(nameof(Function));
|
||||
// logger.LogInformation("Saying hello.");
|
||||
|
||||
// // returns ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
|
||||
// return new List<string>();
|
||||
//}
|
||||
|
||||
//[Function("dafx-Orchestration")]
|
||||
public static async Task<List<string>> RunWorkflowOrchestratorAsync(TaskOrchestrationContext taskOrchestrationContext)
|
||||
{
|
||||
//ILogger logger = context.CreateReplaySafeLogger(nameof(Function));
|
||||
//logger.LogInformation("Invoking RunWorkflowOrchestrator");
|
||||
var outputs = new List<string>();
|
||||
|
||||
await Task.Delay(1);
|
||||
outputs.Add("to do - call get executor result");
|
||||
|
||||
return outputs;
|
||||
}
|
||||
|
||||
// Exposed as an entity trigger via AgentFunctionsProvider
|
||||
public static Task<string> InvokeAgentAsync(
|
||||
[DurableClient] DurableTaskClient client,
|
||||
@@ -43,6 +83,25 @@ internal static class BuiltInFunctions
|
||||
return GrpcEntityRunner.LoadAndRunAsync(encodedEntityRequest, entity, combinedServiceProvider);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Invokes a workflow orchestration in response to an HTTP request.
|
||||
/// </summary>
|
||||
public static async Task<HttpResponseData> RunWorkflowOrechstrtationHttpTriggerAsync(
|
||||
[HttpTrigger] HttpRequestData req,
|
||||
[DurableClient] DurableTaskClient client,
|
||||
FunctionContext context)
|
||||
{
|
||||
// to do: Retrieve the workflow and execute it.
|
||||
var workflowName = context.FunctionDefinition.Name.Replace("http", "dafx");
|
||||
|
||||
//string instanceId = await client.ScheduleNewOrchestrationInstanceAsync("dafx-MyTestWorkflow");
|
||||
string instanceId = await client.ScheduleNewOrchestrationInstanceAsync("OrchFunction"); // dafx-MyTestWorkflow");
|
||||
|
||||
HttpResponseData response = req.CreateResponse(HttpStatusCode.OK);
|
||||
await response.WriteStringAsync($"InvokeWorkflowOrechstrtationAsync is invoked for {workflowName}.{instanceId}");
|
||||
return response;
|
||||
}
|
||||
|
||||
public static async Task<HttpResponseData> RunAgentHttpAsync(
|
||||
[HttpTrigger] HttpRequestData req,
|
||||
[DurableClient] DurableTaskClient client,
|
||||
|
||||
+132
@@ -0,0 +1,132 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.DurableTask;
|
||||
using Microsoft.Azure.Functions.Worker.Core.FunctionMetadata;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Microsoft.Agents.AI.Hosting.AzureFunctions;
|
||||
|
||||
internal sealed class DurableWorkflowFunctionMetadataTransformer : IFunctionMetadataTransformer
|
||||
{
|
||||
private readonly ILogger<DurableWorkflowFunctionMetadataTransformer> _logger;
|
||||
private readonly DurableWorkflowOptions _options;
|
||||
|
||||
public DurableWorkflowFunctionMetadataTransformer(ILogger<DurableWorkflowFunctionMetadataTransformer> logger, DurableWorkflowOptions durableWorkflowOptions)
|
||||
{
|
||||
this._logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
this._options = durableWorkflowOptions ?? throw new ArgumentNullException(nameof(durableWorkflowOptions));
|
||||
}
|
||||
|
||||
public string Name => nameof(DurableWorkflowFunctionMetadataTransformer);
|
||||
|
||||
public void Transform(IList<IFunctionMetadata> original)
|
||||
{
|
||||
if (this._logger.IsEnabled(LogLevel.Information))
|
||||
{
|
||||
this._logger.LogInformation("Transforming function metadata to add durable workflow functions. Initial function count: {FunctionCount}", original.Count);
|
||||
}
|
||||
|
||||
foreach (var workflow in this._options.Workflows)
|
||||
{
|
||||
if (this._logger.IsEnabled(LogLevel.Information))
|
||||
{
|
||||
this._logger.LogInformation("Adding durable workflow function for workflow: {WorkflowName}", workflow.Key);
|
||||
}
|
||||
|
||||
original.Add(CreateOrchestrationTrigger(workflow.Key));
|
||||
// We also want to create an HTTP trigge for this orchestration so users can start it via HTTP.
|
||||
if (this._logger.IsEnabled(LogLevel.Information))
|
||||
{
|
||||
this._logger.LogInformation("Adding HTTP trigger function for workflow: {WorkflowName}", workflow.Key);
|
||||
var httpTriggerMetadata = CreateHttpTrigger(workflow.Key, $"workflows/{workflow.Key}/run");
|
||||
original.Add(httpTriggerMetadata);
|
||||
}
|
||||
|
||||
// Create activity functions for each executor in the workflow
|
||||
// Extract executor IDs from edges and start executor (since ExecutorBindings is internal)
|
||||
var executorIds = new HashSet<string> { workflow.Value.StartExecutorId };
|
||||
|
||||
var reflectedEdges = workflow.Value.ReflectEdges();
|
||||
foreach (var (sourceId, edgeSet) in reflectedEdges)
|
||||
{
|
||||
executorIds.Add(sourceId);
|
||||
foreach (var edge in edgeSet)
|
||||
{
|
||||
foreach (var sinkId in edge.Connection.SinkIds)
|
||||
{
|
||||
executorIds.Add(sinkId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
foreach (var executorId in executorIds)
|
||||
{
|
||||
if (this._logger.IsEnabled(LogLevel.Information))
|
||||
{
|
||||
this._logger.LogInformation(
|
||||
"Adding activity function for executor: {ExecutorId} in workflow: {WorkflowName}",
|
||||
executorId,
|
||||
workflow.Key);
|
||||
}
|
||||
|
||||
original.Add(CreateActivityTrigger(workflow.Key, executorId));
|
||||
}
|
||||
}
|
||||
|
||||
if (this._logger.IsEnabled(LogLevel.Information))
|
||||
{
|
||||
this._logger.LogInformation("Transform finished. Updated function count: {FunctionCount}", original.Count);
|
||||
}
|
||||
}
|
||||
|
||||
private static DefaultFunctionMetadata CreateHttpTrigger(string name, string route)
|
||||
{
|
||||
return new DefaultFunctionMetadata()
|
||||
{
|
||||
Name = $"{BuiltInFunctions.HttpPrefix}{name}",
|
||||
Language = "dotnet-isolated",
|
||||
RawBindings =
|
||||
[
|
||||
$"{{\"name\":\"req\",\"type\":\"httpTrigger\",\"direction\":\"In\",\"authLevel\":\"function\",\"methods\": [\"post\"],\"route\":\"{route}\"}}",
|
||||
"{\"name\":\"$return\",\"type\":\"http\",\"direction\":\"Out\"}",
|
||||
"{\"name\":\"client\",\"type\":\"durableClient\",\"direction\":\"In\"}"
|
||||
],
|
||||
EntryPoint = BuiltInFunctions.RunWorkflowOrechstrtationHttpFunctionEntryPoint,
|
||||
ScriptFile = BuiltInFunctions.ScriptFile
|
||||
};
|
||||
}
|
||||
|
||||
private static DefaultFunctionMetadata CreateOrchestrationTrigger(string name)
|
||||
{
|
||||
return new DefaultFunctionMetadata()
|
||||
{
|
||||
Name = AgentSessionId.ToEntityName(name),
|
||||
Language = "dotnet-isolated",
|
||||
RawBindings =
|
||||
[
|
||||
// """{"name":"context","type":"orchestrationTrigger","direction":"In"}""",
|
||||
"""{"name":"taskOrchestrationContext","type":"orchestrationTrigger","direction":"In"}""",
|
||||
|
||||
],
|
||||
EntryPoint = BuiltInFunctions.RunWorkflowOrechstrtationFunctionEntryPoint,
|
||||
ScriptFile = BuiltInFunctions.ScriptFile,
|
||||
};
|
||||
}
|
||||
|
||||
private static DefaultFunctionMetadata CreateActivityTrigger(string workflowName, string executorId)
|
||||
{
|
||||
string functionName = $"{AgentSessionId.ToEntityName(workflowName)}_{executorId}";
|
||||
|
||||
return new DefaultFunctionMetadata()
|
||||
{
|
||||
Name = functionName,
|
||||
Language = "dotnet-isolated",
|
||||
RawBindings =
|
||||
[
|
||||
"""{"name":"input","type":"activityTrigger","direction":"In","dataType":"String"}""",
|
||||
],
|
||||
EntryPoint = BuiltInFunctions.InvokeWorkflowActivityFunctionEntryPoint,
|
||||
ScriptFile = BuiltInFunctions.ScriptFile,
|
||||
};
|
||||
}
|
||||
}
|
||||
+18
@@ -14,6 +14,22 @@ namespace Microsoft.Agents.AI.Hosting.AzureFunctions;
|
||||
/// </summary>
|
||||
public static class FunctionsApplicationBuilderExtensions
|
||||
{
|
||||
/// <summary>
|
||||
/// Adds support for durable workflows to the specified Functions application builder.
|
||||
/// </summary>
|
||||
/// <param name="builder">The Functions application builder to configure with durable workflow capabilities.</param>
|
||||
/// <param name="configure"></param>
|
||||
/// <returns>The same instance of <see cref="FunctionsApplicationBuilder"/> to allow for method chaining.</returns>
|
||||
public static FunctionsApplicationBuilder AddDurableWorkflows(this FunctionsApplicationBuilder builder, Action<DurableWorkflowOptions> configure)
|
||||
{
|
||||
var options = new DurableWorkflowOptions();
|
||||
configure(options);
|
||||
builder.Services.AddSingleton(options);
|
||||
builder.Services.AddSingleton<IFunctionMetadataTransformer, DurableWorkflowFunctionMetadataTransformer>();
|
||||
|
||||
return builder;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Configures the application to use durable agents with a builder pattern.
|
||||
/// </summary>
|
||||
@@ -38,6 +54,8 @@ public static class FunctionsApplicationBuilderExtensions
|
||||
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.RunWorkflowOrechstrtationFunctionEntryPoint, StringComparison.Ordinal) ||
|
||||
string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunWorkflowOrechstrtationHttpFunctionEntryPoint, StringComparison.Ordinal) ||
|
||||
string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentEntityFunctionEntryPoint, StringComparison.Ordinal));
|
||||
builder.Services.AddSingleton<BuiltInFunctionExecutor>();
|
||||
|
||||
|
||||
Reference in New Issue
Block a user