Workflow Orchestration Patterns in Microsoft Agent Framework
My previous article focused on introduction of workflows in MAF along with a basic overview of Executors, Edges and Events.
This article will explore different workflow orchestration patterns available in MAF. Orchestration patterns help manage and coordinate workflow execution in specific ways depending on the use case.
The available orchestration types available in MAF are:
Sequential : They execute one after another in a defined order.
Concurrent : The agents in this pattern execute in parallel .
Handoff : Here the agents transfer control to each other based on context.
Group Chat : The agents in group chat collaborate in a shared conversation
Magnetic : A manager agent dynamically coordinates specialized agents
Of the above patterns, Magnetic pattern is not supported in C# , so this article will skip that and focus on the remaining four.
In MAF when an AIAgent is passed to WorkflowBuilder, MAF automatically wraps it in an AIAgentBinding which creates the underlying AIAgentHostExecutor.
The workflow output can be streamed dynamically through.
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
SetUp
Create a new C# console application and add the following packages
dotnet add package Azure;
dotnet add package Azure.AI.OpenAI;
dotnet add package Microsoft.Agents.AI;
dotnet add package Microsoft.Extensions.AI;
dotnet add package Microsoft.Extensions.Configuration;
dotnet add package Microsoft.Extensions.DependencyInjection;
dotnet add package Microsoft.Extensions.Logging;
dotnet add package Microsoft.Agents.AI.Workflows;
Add appsetting.json to the project
"AppSettings": {
"Chat_DeploymentName": "Deployment Name",
"EndPoint": "Azure OpenAI endpoint",
"ApiKey": "Azure OpenAI API key"
}
Code
Now that we have all the underlying artifacts in place, add the following code to read the settings from appsettings.json
var configuration = new ConfigurationBuilder()
.SetBasePath(Directory.GetCurrentDirectory())
.AddJsonFile("appsettings.json", optional: false)
.Build();
Create a DI service registry.
ServiceCollection servicecollection = new ServiceCollection();
Now that we have the basic set up ready, lets create specific agents for different workflow orchestration patterns.
Sequential Orchestration Pattern
Lets assume, we have a translation agent that translates given text into different languages. In our example the languages are French, German and Spanish.
Below we set up a translation agent that translates a given text from English into desired language/s.
var credential = new AzureKeyCredential(configuration["AppSettings:ApiKey"]);
servicecollection.AddKeyedChatClient("ChatClient", (sp => new AzureOpenAIClient(
new Uri(configuration["AppSettings:EndPoint"]), credential)
.GetChatClient(configuration["AppSettings:Chat_DeploymentName"])
.AsIChatClient()));
servicecollection.AddSingleton < AIAgent > (sp =>
{
Func < ChatClientAgentOptions > func = () => {
return new ChatClientAgentOptions {
ChatOptions = new ChatOptions {
Instructions = "You are a translation agent.You translate text from English to another language.Translate the value in 'Text' into the language specified in 'Target Language'.Requires parameter:Text (string), Target Language (string).Return only the translated text. ",
},
Name = "TranslationAgent",
Id = "1"
};
};
return new ChatClientAgent(sp.GetKeyedService < IChatClient > ("ChatClient"), options: func());
});
We create a Dependency Injection (DI) container serviceProvider from the registered services in serviceCollection.
ServiceProvider serviceProvider = servicecollection.BuildServiceProvider();
We specify a list of languages along with the text that requires to be translated.
The agent executor in C# supports multiple input types, including
string,ChatMessage, andIEnumerable<ChatMessage>. When astringis provided as input, it is automatically converted into aChatMessagewith theUserrole.
So the input query can either be
string messages_input = $"Target Language: \"French\", \"German\", \"Spanish\" and Text: Hello ";
Or
List<ChatMessage> messages_input = [
new(ChatRole.System, $"Target Language: \"French\", \"German\", \"Spanish\" and Text: Hello")];
Note: Above we have specified Target Language and Text variables in the Agent instructions.
We resolve the registered agent from the DI container and set the collection of type List<AIAgent> .
List<AIAgent> agent = new(serviceProvider.GetServices<AIAgent>());
Next, we declare list of type List<Messages> to capture and store the workflow output.
List<ChatMessage> messages_output = [];
We then initiate asynchronous streaming execution of the workflow and also set the workflow execution pattern type . In this example it is Sequential.
await using StreamingRun run = await InProcessExecution.RunStreamingAsync(AgentWorkflowBuilder.BuildSequential(agent), messages_input);
We can use the Events listed here to trace the output across each stage.
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
await foreach(WorkflowEvent evt in run.WatchStreamAsync())
{
if (evt is AgentResponseUpdateEvent update) {
AgentResponse response = update.AsResponse();
foreach(ChatMessage message in response.Messages.Where(a => a.Role.Value == "assistant" && a.Text != "")) {
messages_output.Add(new(ChatRole.Assistant, message.Text));
}
}
else if (evt is WorkflowErrorEvent workflowError) {
Console.Error.WriteLine(workflowError.Exception?.ToString() ?? "Unknown workflow error occurred.");
}
else if (evt is ExecutorFailedEvent executorFailed) {
Console.Error.WriteLine($"Executor '{executorFailed.ExecutorId}' failed with {(executorFailed.Data == null ? "
unknown error " : $"
exception {executorFailed.Data}")}.");
}
}
Console.WriteLine(string.Join("", messages_output.Where(a => a.Role.Value == "assistant").Select(a => a.Text))
In the above code, the agent processes its cached messages only after receiving a TurnToken.
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
If TurnToken is not sent the AIAgentHostExecutor will not process the workflow.
Execution >>
Concurrent Orchestration Pattern
For Concurrent pattern , the complete code stays the same as the Sequential pattern. The only change required is to change AgentWorkflowBuilder type from BuildSequential to BuildConcurrent.
await using StreamingRun run = await InProcessExecution.RunStreamingAsync(AgentWorkflowBuilder.BuildConcurrent(agent), messages_input);
HandOff Orchestration Pattern
HandOff orchestration at first glance might look similar to agents-as-tools but there is a subtle difference between the two.
In agent-as-tools you basically use an agent as tool for another agent through Agent.AsAIFunction() which performs the execution on behalf of the calling agent.
For example , below the AnotherAgentperforms call to the StockExchangeAgentas agent-as-tool through Agent.AsAIFunction().
StockExchangeAgent =......
.AsAIAgent(
model: "your model",
instructions: "You answer questions about stocks",
name: "StockExchangeAgent",
description: "An agent that answers questions about the stocks.",
tools: [AIFunctionFactory.Create(GetStockvalue)]);
AnotherAgent =......
.AsAIAgent(
model: "your model",
instructions: "You are a helpful assistant.",
tools: [StockExchangeAgent.AsAIFunction()]);
But in Hand-Off pattern, the agent receiving the hand-off takes complete responsibility of the entire execution.
An agent dynamically transfers control to another agent after completing its task.
To put in perspective, a handoff occurs when one agent passes responsibility, conversational context and execution flow to another agent to maintain continuity of the execution flow.
Lets look at an sample scenario. Assume we have three agents
Math Agent : That answers ONLY Math questions.
Chemistry Agent : That answers ONLY Chemistry questions.
Coordinator Agent : Has access to the above two specialized agents and acts as a coordinator and is aware of its role in the workflow and agents that it has at its disposal.
We have to ensure that all the above agents are aware of their roles and responsibilities during the execution. This can be implemented by setting relevant instructions for every agent when setting them up.
Moderator Agent >>
servicecollection.AddSingleton<AIAgent>(sp =>
{
Func<ChatClientAgentOptions> func = () =>
{
return new ChatClientAgentOptions
{
ChatOptions = new ChatOptions
{
Instructions = "You determine which agent to use based on the user questions. ALWAYS handoff to another agent and do not try to respond by yourself. If you don't find any relevant agent for a specific user question inform the user that the question is irrelevant."
},
Name = "ModeratorAgent",
Id = "1"
};
};
return new ChatClientAgent(sp.GetKeyedService<IChatClient>("ChatClient"), options: func());
});
Math Agent >>
servicecollection.AddSingleton<AIAgent>(sp =>
{
Func<ChatClientAgentOptions> func = () =>
{
return new ChatClientAgentOptions
{
ChatOptions = new ChatOptions
{
Instructions = "You provide help with math problems. Explain your reasoning at each step and include examples. Only respond about math queries."
},
Name = "MathAgent",
Id = "2"
};
};
return new ChatClientAgent(sp.GetKeyedService<IChatClient>("ChatClient"), options: func());
});
Chemistry Agent >>
servicecollection.AddSingleton<AIAgent>(sp =>
{
Func<ChatClientAgentOptions> func = () =>
{
return new ChatClientAgentOptions
{
ChatOptions = new ChatOptions
{
Instructions = "You provide help with chemistry problems. Explain your reasoning at each step and include examples. Only respond about chemistry queries."
},
Name = "ChemistryAgent",
Id = "3"
};
};
return new ChatClientAgent(sp.GetKeyedService<IChatClient>("ChatClient"), options: func());
});
Once the agents are defined, we have to set up the workflow but before that we fetch the list of agents from the DI container
var agents = serviceProvider.GetServices<AIAgent>();
List<AIAgent> AgentList = new(agents);
We then define the workflow
#pragma warning disable MAAIW001 // Type is for evaluation purposes only and is subject to change or removal in future updates. Suppress this diagnostic to proceed.
var workflow = AgentWorkflowBuilder.CreateHandoffBuilderWith(AgentList[0])
.WithHandoffs(AgentList[0], [AgentList[1], AgentList[2]])
.WithHandoffs([AgentList[1], AgentList[2]], AgentList[0])
.Build();
#pragma warning restore MAAIW001 // Type is for evaluation purposes only and is subject to change or removal in future updates. Suppress this diagnostic to proceed.
Since handoff workflow feature is in preview, ensure that you have set the #pragma warning MAAIW001
The AgentList[index] above is derived from the agents collection.
In the next step , we ask a series of questions. One related to Math, one related to Chemistry and the last one related to History.
We declare a variable called lastExecutorId to differentiate responses between individual executors.
The user queries are passed through List<ChatMessage> messages = []; while the chat history is stored in List<ChatMessage> messages_output = [];.This is optional incase the chat history is required to be stored for further analysis. You can refer to my this article on storing chat history in MAF.
We pass a series of questions to the agents.
List<ChatMessage> messages = [];
List<ChatMessage> messages_output = [];
messages.Add(new(ChatRole.User, "What is a derivative?"));
messages.Add(new(ChatRole.User, "What is the Chemical formula for water?"));
messages.Add(new(ChatRole.User, "Tell me more about Middle ages"));
string? lastExecutorId = null;
and set up the workflow builder.
foreach (var msg in messages)
{
await using StreamingRun run_ = await InProcessExecution.RunStreamingAsync(workflow, msg);
messages_output.Add(new(ChatRole.User, msg.Text));
await run_.TrySendMessageAsync(new TurnToken(emitEvents: true));
await foreach (WorkflowEvent evt in run_.WatchStreamAsync())
{
if (evt is AgentResponseUpdateEvent e)
{
if (e.ExecutorId != lastExecutorId)
{
Console.WriteLine();
lastExecutorId = e.ExecutorId;
Console.WriteLine();
}
Console.Write(e.Update.Text);
messages_output.Add(new(ChatRole.Assistant, e.Update.Text));
if (e.Update.Contents.OfType<FunctionCallContent>().FirstOrDefault() is FunctionCallContent call)
{
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine("---------------------------------------------------------------");
Console.WriteLine($" [Calling function '{call.Name}' with arguments: {JsonSerializer.Serialize(call.Arguments)}]");
Console.WriteLine("The user query is : " + msg);
Console.WriteLine("---------------------------------------------------------------");
Console.ResetColor();
}
}
}
}
Execution >>
Group Chat Orchestration Pattern
This is a collaborative pattern where an orchestrator decides the conversation flow. Unlike the handoff pattern where the moderator was an agent, the orchestrator in group chat pattern is not an agent.
The agents are selected by the orchestrator to coordinate who executes next. The agents here have an advantage where they can review other agents responses in multiple iterations.
The orchestrator can use different strategies like round-robin, prompt-based, custom logic to select speakers.
Lets assume a use case where we define an agent that creates slogan for a football club and the second agent which is a reviewer, reviews the created content and provides feedback and suggests improvements.
ContentCreatorAgent >>
servicecollection.AddSingleton<AIAgent>(sp =>
{
Func<ChatClientAgentOptions> func = () =>
{
return new ChatClientAgentOptions()
{
ChatOptions = new ChatOptions
{
Instructions = "You are a content creator.You create content for football teams.Be concise and please stick to the topic."
},
Name = "ContentCreator",
Id = "1"
};
};
return new ChatClientAgent(sp.GetKeyedService<IChatClient>("ChatClient"), options: func());
}
);
ContentReviewerAgent >>
servicecollection.AddSingleton<AIAgent>(sp =>
{
Func<ChatClientAgentOptions> func = () =>
{
return new ChatClientAgentOptions()
{
ChatOptions = new ChatOptions
{
Instructions = "You are a content reviewer.You review content for football teams.Be concise and please stick to the topic."
},
Name = "ContentReviewer",
Id = "2"
};
};
return new ChatClientAgent(sp.GetKeyedService<IChatClient>("ChatClient"), options: func());
}
);
Once the agents are defined, we have to set up the workflow but before that ,we fetch the list of agents from the DI container.
var agents = serviceProvider.GetServices<AIAgent>();
List<AIAgent> AgentList = new(agents);
We then define the workflow
var workflowcontent = AgentWorkflowBuilder
.CreateGroupChatBuilderWith(agents =>
new RoundRobinGroupChatManager(agents)
{
MaximumIterationCount = 5 // Maximum number of turns
})
.AddParticipants(AgentList[0], AgentList[1])
.Build();
We then set up the workflow and also add the entire conversation to messages that is of typeList<ChatMessages>required only incase we need to store the entire conversation.
var messages = new List < ChatMessage >
{
new(ChatRole.User, "Create a slogan for football club FC Barcelona")
};
string ? lastExecutorId = null;
await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflowcontent, messages);
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
await foreach(WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
{
if (evt is AgentResponseUpdateEvent e)
{
if (e.ExecutorId != lastExecutorId)
{
Console.WriteLine();
lastExecutorId = e.ExecutorId;
Console.WriteLine(e.ExecutorId);
Console.WriteLine();
}
Console.Write(e.Update.Text);
messages.Add(new(ChatRole.Assistant, e.Update.Text));
}
}
Execution >>
Wrapping It Up
Selection of workflow orchestration pattern in MAF is very crucial mechanism to make multi agent system to work effectively and efficiently. In this article I tried covering the most important workflow orchestration pattern provided in MAF.
Irrespective of the underlying framework used, the idea remains the same: clear, precise and consistent selection of workflow orchestration pattern enable agents to work together effectively.
Thanks for reading !!!
