Human In the Loop(HITL) in Azure Durable Functions for Microsoft Agent Framework
My previous article focused on Multi Agent Orchestration in Azure Durable Functions for Microsoft Agent Framework.
The use case used in the article was pretty straightforward wherein there were two agents .
Content generating agent
Content review agent
Content generated by the first agent is reviewed, modified and pushed forward for further processing by the second agent.
But such an straightforward approach might carry potential risks and might not be feasible in real life scenarios particularly for crucial business processes that adhere to strict compliances.
We wouldn't want agents to work autonomously and take decisions without proper human reviews . This is where Human In The Loop (HITL) plays an important role.
This article focusses on introducing Human In The Loop (HITL) concept for Azure Durable functions in MAF.
Also in the previous article, the user input was embedded in the code and there wasn't a mechanism to make the input dynamic . This article covers that aspect as well where the user input is dynamic and is sent while REST API invocation.
If you would like to skip the writeup you can view the flow walkthrough here
Flow >>
The flow is pretty straightforward.
The entry point is an HTTP Trigger that accepts the input as HttpRequestData and creates an orchestration instance that invokes an Orchestration method called RunOrchestrationAsync.
In RunOrchestrationAsync method, the agent creates the content on the topic provided and a notification is sent to the user(human) to approve or reject it. In the meantime the flow pauses for the human response.
Once the RunOrchestrationAsync receives the response from the user(human), the method invokes the PublishContent activity if approved else invokes the NotifyUseForRejection activity if rejected . This concludes the orchestration execution.
SetUp
To get started, create a new Azure Function project and apply settings covered in my previous article including setting up of the Docker DTS Emulator.
Ensure Docker DTS Emulator is up and navigate to http://localhost:8082/ to ensure that it is running.
Code
After the above artifacts are in place, add the following code to read the settings from appsettings.json in Program.cs of the project.
var configuration = new ConfigurationBuilder()
.SetBasePath(Directory.GetCurrentDirectory())
.AddJsonFile("appsettings.json", optional: false)
.Build();
Read the credentials and register a Chatclient
var credential = new AzureKeyCredential(configuration["AppSettings:ApiKey"]);
ServiceCollection servicecollection = new();
builder.Services.AddKeyedChatClient(
"ChatClient",
(
sp =>
new AzureOpenAIClient(new Uri(configuration["AppSettings:EndPoint"]), credential)
.GetChatClient(configuration["AppSettings:Chat_DeploymentName"])
.AsIChatClient()
)
);
In the next step, register the following agent in the hosted DI container.
FootballContentCreatorAgent >>
servicecollection.AddSingleton<ChatClientAgent>(sp =>
{
Func<ChatClientAgentOptions> func = () =>
{
return new ChatClientAgentOptions
{
ChatOptions = new ChatOptions
{
Instructions = "You are a content creator.You are good at writing reviews for football club.Be concise and please stick to the topic.",
},
Name = "FootballContentCreatorAgent",
Id = "1"
};
};
return new ChatClientAgent(sp.GetKeyedService<IChatClient>("ChatClient"), options: func());
});
Build the service and fetch agents from the ServiceProvider as ChatClientAgent.
ServiceProvider serviceProvider = servicecollection.BuildServiceProvider();
var agent = serviceProvider.GetServices<ChatClientAgent>();
List<ChatClientAgents> chatclientagent = new(agent);
Then add these agents as DurableAgent to the Azure Function worker.
using IHost app = FunctionsApplication
.CreateBuilder(args)
.ConfigureFunctionsWebApplication()
.ConfigureDurableAgents(options => options.AddAIAgent(chatclientagent[0], timeToLive: TimeSpan.FromHours(1)))
.Build();
app.Run();
Create a new class file called FunctionTigger.cs and add two classes Input and HumanApproval.
Input >>
public class Input
{
[JsonPropertyName("input")]
public string input { get; set; }
}
HumanApproval>>
public class HumanApproval
{
[JsonPropertyName("IsApproved")]
public string IsApproved { get; set; }
}
Note : I have kept the structure pretty simple. You can add additional properties based on your use case. For ex , a Feedback property in the HumanApproval
class that accepts feedback from the human(user) prior to an approval/rejection.
Also add a Record object called TextResponse that accepts AgentResponses as input.
public record TextResponse(string Response);
StartOrchestrationAsync >>
As shown in the flowchart, StartOrchestrationAsync will be the method that acts as an entry point when the AzureFunction is triggered through an POST request to the HTTP endpoint.
[Function(nameof(StartOrchestrationAsync))]
public static async Task<HttpResponseData> StartOrchestrationAsync(
[HttpTrigger(AuthorizationLevel.Anonymous, "post", Route = "hitl/run")] HttpRequestData req,
[DurableClient] DurableTaskClient client)
{
var input = await req.ReadFromJsonAsync<Input>();
string instanceid = await client.ScheduleNewOrchestrationInstanceAsync(orchestratorName: nameof(RunOrchestrationAsync), input: input);
var response = req.CreateResponse(System.Net.HttpStatusCode.Accepted);
return response;
}
The endpoint for the orchestration in the above example is "hitl/run".
A new orchestration instance is created in this method and which in turn invokes the RunOrchestrationAsync method.
RunOrchestrationAsync >>
[Function(nameof(RunOrchestrationAsync))]
public static async Task RunOrchestrationAsync([OrchestrationTrigger] TaskOrchestrationContext context)
{
DurableAIAgent FootballContentCreatorAgent = context.GetAgent("FootballContentCreatorAgent");
var input = context.GetInput<Input>();
AgentSession Session = await FootballContentCreatorAgent.CreateSessionAsync();
AgentResponse<TextResponse> content = await FootballContentCreatorAgent.RunAsync<TextResponse>(
message: input.input,
session: Session);
TextResponse agentresponse = content.Result;
await context.CallActivityAsync(nameof(NotifyUserForApproval), agentresponse);
HumanApproval humanResponse;
humanResponse = await context.WaitForExternalEvent<HumanApproval>(
eventName: "HumanApproval",
timeout: TimeSpan.FromHours(1));
if (humanResponse.IsApproved == "Yes")
{
await context.CallActivityAsync(nameof(PublishContent), agentresponse);
context.SetCustomStatus($"Content published successfully at {context.CurrentUtcDateTime.ToLongTimeString}");
}
else
{
await context.CallActivityAsync(nameof(NotifyUserForRejection), agentresponse);
context.SetCustomStatus("Content is rejected by human reviewer. Publishing content...");
}
}
RunOrchestrationAsync is method is the most important method that processes the user input.
Lets breakdown the above method :
We first create a new instance of DurableAIAgent from the context . The user input from the http endpoint is stored in the input variable of type T declared earlier.
DurableAIAgent FootballContentCreatorAgent = context.GetAgent("FootballContentCreatorAgent");
var input = context.GetInput<Input>();
Create a new AgentSession. Executing the agent returns an output of type AgentResponse.
Then retrieve the underlying TextResponse from the Result property into the variable agentresponse.
AgentSession Session = await FootballContentCreatorAgent.CreateSessionAsync();
AgentResponse<TextResponse> content = await FootballContentCreatorAgent.RunAsync<TextResponse>(
message: input.input,
session: Session);
TextResponse agentresponse = content.Result;
Execute CallActivityAsync that invokes a method named NotifyUserForApproval with agentresponse created above as the parameter to the method.
await context.CallActivityAsync(nameof(NotifyUserForApproval), agentresponse);
Next, the method waits for humanapproval event through WaitForExternalEvent for a timespan of one hour.
HumanApproval humanResponse = await context.WaitForExternalEvent<HumanApproval>(
eventName: "HumanApproval",
timeout: TimeSpan.FromHours(1));
Once the approval is approved/rejected the underlying methods are invoked through CallActivityAsync based on the type of input received.
if (humanResponse.IsApproved == "Yes")
{
await context.CallActivityAsync(nameof(PublishContent), agentresponse);
context.SetCustomStatus($"Content published successfully at {context.CurrentUtcDateTime.ToLongTimeString}");
}
else
{
await context.CallActivityAsync(nameof(NotifyUserForRejection), agentresponse);
context.SetCustomStatus("Content is rejected by human reviewer. Publishing content...");
}
NotifyUserForApproval >>
In real life use case an email or other forms of notifications with an link to the endpoint should be sent to for approval or rejections. This link is generated from the HumanApprovalAsync method explained later in the execution flow.
[Function(nameof(NotifyUserForApproval))]
public async static Task<string> NotifyUserForApproval(
[ActivityTrigger] TextResponse content,
FunctionContext functionContext)
{
return $"Please review the following generated content{content}";
}
PublishContent >>
[Function(nameof(PublishContent))]
public async static Task<string> PublishContent(
[ActivityTrigger] TextResponse content,
FunctionContext functionContext)
{
return $"The following content {content} is approved and is published";
}
NotifyUserForRejection>>
[Function(nameof(NotifyUserForRejection))]
public async static Task<string> NotifyUserForRejection(
[ActivityTrigger] TextResponse content,
FunctionContext functionContext)
{
return $"The following content {content} is rejected ";
}
HumanApprovalAsync>>
The approval/rejection endpoint with hitl/notification/{instanceId} is created in HumanApprovalAsync method and a response is returned with IsApproved value set based on the user input.
[Function(nameof(HumanApprovalAsync))]
public static async Task<HttpResponseData> HumanApprovalAsync(
[HttpTrigger(AuthorizationLevel.Anonymous, "post", Route = "hitl/notification/{instanceId}")] HttpRequestData req, string instanceId,
[DurableClient] DurableTaskClient client)
{
var humanapproval = await req.ReadFromJsonAsync<HumanApproval>();
await client.RaiseEventAsync(instanceId, "HumanApproval", humanapproval);
HttpResponseData response = req.CreateResponse(HttpStatusCode.OK);
return response;
}
You might ask how would the user approve/reject the request.
This can be done through the HTTPost that invokes the notification endpoints created in the HumanApprovalAsync method. Below is the PowerShell call to the endpoints.
HTTP POST request to the StartOrchestrationAsync method >>
$body = @{
input = "Write a review for the football club FC Barcelona"
} | ConvertTo-Json
Invoke-RestMethod -Method Post `
-Uri http://localhost:{Port no set in your launchsettings.json file}/api/hitl/run `
-ContentType application/json `
-Body $body
HTTP POST request to the HumanApprovalAsync method >>
$json = '{"IsApproved":"Yes"}'
Invoke-RestMethod `
-Uri "http://localhost:{Port no set in your launchsettings.json file}/api/hitl/notification/{InstanceId from StartOrchestrationAsync method}" `
-Method Post `
-ContentType "application/json" `
-Body $json
DTS Dashboard for User Approval >>
DTS Dashboard for User Rejection >>
Execution
Conclusion
Through this article, I tried to explore a simple implementation of Human-in-the-Loop (HITL) for Azure Durable Functions.
Although the sample focuses on a basic approval workflow it highlights the core concepts required to pause an orchestration, wait for human input, and resume execution based on the outcome.
I hope this article helps you understand the fundamentals and sets the required foundation for building more business centric HITL solutions in your applications.
Thanks for reading !!!
