mirror of
https://github.com/davidkaya/aryx.git
synced 2026-08-27 21:33:58 +02:00
refactor: pin workflow host options explicitly
Explicitly configure AIAgentHostOptions for the workflow modes that host agents directly instead of relying on Agent Framework defaults. - add a shared host-options factory that preserves Aryx's current behavior - use custom sequential, concurrent, and round-robin group-chat builders so all host options are set intentionally - add workflow-level tests asserting the configured host options Keep EmitAgentResponseEvents disabled because Aryx still projects streaming transcript state itself and enabling response events would require a separate reconciliation change. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
David Kaya
@@ -189,9 +189,9 @@ internal sealed class CopilotAgentBundle : IAsyncDisposable
|
||||
{
|
||||
return pattern.Mode switch
|
||||
{
|
||||
"single" => AgentWorkflowBuilder.BuildSequential(pattern.Name, ResolveOrderedAgents(pattern)),
|
||||
"sequential" => AgentWorkflowBuilder.BuildSequential(pattern.Name, ResolveOrderedAgents(pattern)),
|
||||
"concurrent" => AgentWorkflowBuilder.BuildConcurrent(pattern.Name, ResolveOrderedAgents(pattern)),
|
||||
"single" => BuildSequentialWorkflow(pattern),
|
||||
"sequential" => BuildSequentialWorkflow(pattern),
|
||||
"concurrent" => BuildConcurrentWorkflow(pattern),
|
||||
"handoff" => BuildHandoffWorkflow(pattern),
|
||||
"group-chat" => BuildGroupChatWorkflow(pattern),
|
||||
"magentic" => throw new NotSupportedException(
|
||||
@@ -249,6 +249,23 @@ internal sealed class CopilotAgentBundle : IAsyncDisposable
|
||||
return builder.Build();
|
||||
}
|
||||
|
||||
internal static AIAgentHostOptions CreateAgentHostOptions()
|
||||
{
|
||||
return new AIAgentHostOptions
|
||||
{
|
||||
// Aryx controls per-turn streaming with TurnToken(emitEvents: true), so keep this
|
||||
// null to preserve that behavior while making the host defaults explicit in code.
|
||||
EmitAgentUpdateEvents = null,
|
||||
// Aryx already projects streamed transcript state itself; enabling this would add
|
||||
// extra response events that need separate reconciliation first.
|
||||
EmitAgentResponseEvents = false,
|
||||
InterceptUserInputRequests = false,
|
||||
InterceptUnterminatedFunctionCalls = false,
|
||||
ReassignOtherAgentsAsUsers = true,
|
||||
ForwardIncomingMessages = true,
|
||||
};
|
||||
}
|
||||
|
||||
internal static HandoffsWorkflowBuilder CreateHandoffWorkflowBuilder(AIAgent entryAgent)
|
||||
{
|
||||
return AgentWorkflowBuilder.CreateHandoffBuilderWith(entryAgent)
|
||||
@@ -259,20 +276,109 @@ internal sealed class CopilotAgentBundle : IAsyncDisposable
|
||||
.WithHandoffInstructions(HandoffWorkflowGuidance.CreateWorkflowInstructions());
|
||||
}
|
||||
|
||||
private Workflow BuildSequentialWorkflow(PatternDefinitionDto pattern)
|
||||
{
|
||||
IReadOnlyList<AIAgent> agents = ResolveOrderedAgents(pattern);
|
||||
List<ExecutorBinding> agentExecutors = agents
|
||||
.Select(CreateAgentExecutorBinding)
|
||||
.ToList();
|
||||
|
||||
ExecutorBinding previous = agentExecutors[0];
|
||||
WorkflowBuilder builder = new(previous);
|
||||
|
||||
foreach (ExecutorBinding next in agentExecutors.Skip(1))
|
||||
{
|
||||
builder.AddEdge(previous, next);
|
||||
previous = next;
|
||||
}
|
||||
|
||||
WorkflowOutputMessagesExecutor end = new();
|
||||
builder = builder.AddEdge(previous, end).WithOutputFrom(end);
|
||||
|
||||
if (pattern.Name is not null)
|
||||
{
|
||||
builder = builder.WithName(pattern.Name);
|
||||
}
|
||||
|
||||
return builder.Build();
|
||||
}
|
||||
|
||||
private Workflow BuildConcurrentWorkflow(PatternDefinitionDto pattern)
|
||||
{
|
||||
IReadOnlyList<AIAgent> agents = ResolveOrderedAgents(pattern);
|
||||
ChatForwardingExecutor start = new("Start");
|
||||
WorkflowBuilder builder = new(start);
|
||||
|
||||
ExecutorBinding[] agentExecutors = agents
|
||||
.Select(CreateAgentExecutorBinding)
|
||||
.ToArray();
|
||||
ExecutorBinding[] accumulators = agentExecutors
|
||||
.Select(executor => CreateAggregateMessagesExecutorBinding($"Batcher/{executor.Id}"))
|
||||
.ToArray();
|
||||
|
||||
builder.AddFanOutEdge(start, agentExecutors);
|
||||
|
||||
for (int index = 0; index < agentExecutors.Length; index++)
|
||||
{
|
||||
builder.AddEdge(agentExecutors[index], accumulators[index]);
|
||||
}
|
||||
|
||||
Func<string, string, ValueTask<WorkflowConcurrentEndExecutor>> endFactory =
|
||||
(_, __) => new(new WorkflowConcurrentEndExecutor(agentExecutors.Length, AggregateConcurrentResults));
|
||||
ExecutorBinding end = endFactory.BindExecutor(WorkflowConcurrentEndExecutor.ExecutorId);
|
||||
|
||||
builder.AddFanInBarrierEdge(accumulators, end);
|
||||
builder = builder.WithOutputFrom(end);
|
||||
|
||||
if (pattern.Name is not null)
|
||||
{
|
||||
builder = builder.WithName(pattern.Name);
|
||||
}
|
||||
|
||||
return builder.Build();
|
||||
}
|
||||
|
||||
private Workflow BuildGroupChatWorkflow(PatternDefinitionDto pattern)
|
||||
{
|
||||
int maximumIterations = pattern.MaxIterations <= 0 ? 5 : pattern.MaxIterations;
|
||||
AIAgent[] agents = ResolveOrderedAgents(pattern).ToArray();
|
||||
Dictionary<AIAgent, ExecutorBinding> agentMap = agents.ToDictionary(
|
||||
agent => agent,
|
||||
CreateAgentExecutorBinding);
|
||||
|
||||
return AgentWorkflowBuilder
|
||||
.CreateGroupChatBuilderWith(agents =>
|
||||
new RoundRobinGroupChatManager(agents)
|
||||
{
|
||||
MaximumIterationCount = maximumIterations,
|
||||
})
|
||||
.AddParticipants(ResolveOrderedAgents(pattern).ToArray())
|
||||
.Build();
|
||||
Func<string, string, ValueTask<WorkflowRoundRobinGroupChatHost>> groupChatHostFactory =
|
||||
(id, _) => new(new WorkflowRoundRobinGroupChatHost(
|
||||
id,
|
||||
agents,
|
||||
agentMap,
|
||||
maximumIterations));
|
||||
|
||||
ExecutorBinding host = groupChatHostFactory.BindExecutor("GroupChatHost");
|
||||
WorkflowBuilder builder = new(host);
|
||||
|
||||
foreach (ExecutorBinding participant in agentMap.Values)
|
||||
{
|
||||
builder
|
||||
.AddEdge(host, participant)
|
||||
.AddEdge(participant, host);
|
||||
}
|
||||
|
||||
return builder.WithOutputFrom(host).Build();
|
||||
}
|
||||
|
||||
private static ExecutorBinding CreateAgentExecutorBinding(AIAgent agent)
|
||||
=> agent.BindAsExecutor(CreateAgentHostOptions());
|
||||
|
||||
private static ExecutorBinding CreateAggregateMessagesExecutorBinding(string id)
|
||||
{
|
||||
Func<string, string, ValueTask<WorkflowAggregateTurnMessagesExecutor>> factory =
|
||||
(_, __) => new(new WorkflowAggregateTurnMessagesExecutor(id));
|
||||
return factory.BindExecutor(id);
|
||||
}
|
||||
|
||||
private static List<ChatMessage> AggregateConcurrentResults(IList<List<ChatMessage>> lists)
|
||||
=> [.. from list in lists where list.Count > 0 select list.Last()];
|
||||
|
||||
private IReadOnlyList<AIAgent> ResolveOrderedAgents(PatternDefinitionDto pattern)
|
||||
{
|
||||
Dictionary<string, AIAgent> agentMap = BuildAgentMap(pattern);
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
using Microsoft.Agents.AI;
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
using Microsoft.Extensions.AI;
|
||||
|
||||
namespace Aryx.AgentHost.Services;
|
||||
|
||||
internal sealed class WorkflowOutputMessagesExecutor(ChatProtocolExecutorOptions? options = null)
|
||||
: ChatProtocolExecutor(ExecutorId, options, declareCrossRunShareable: true), IResettableExecutor
|
||||
{
|
||||
public const string ExecutorId = "OutputMessages";
|
||||
|
||||
protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)
|
||||
=> base.ConfigureProtocol(protocolBuilder)
|
||||
.YieldsOutput<List<ChatMessage>>();
|
||||
|
||||
protected override ValueTask TakeTurnAsync(
|
||||
List<ChatMessage> messages,
|
||||
IWorkflowContext context,
|
||||
bool? emitEvents,
|
||||
CancellationToken cancellationToken = default)
|
||||
=> context.YieldOutputAsync(messages, cancellationToken);
|
||||
|
||||
ValueTask IResettableExecutor.ResetAsync() => default;
|
||||
}
|
||||
|
||||
internal sealed class WorkflowAggregateTurnMessagesExecutor(string id)
|
||||
: ChatProtocolExecutor(id, s_options, declareCrossRunShareable: true), IResettableExecutor
|
||||
{
|
||||
private static readonly ChatProtocolExecutorOptions s_options = new() { AutoSendTurnToken = false };
|
||||
|
||||
protected override ValueTask TakeTurnAsync(
|
||||
List<ChatMessage> messages,
|
||||
IWorkflowContext context,
|
||||
bool? emitEvents,
|
||||
CancellationToken cancellationToken = default)
|
||||
=> context.SendMessageAsync(messages, cancellationToken: cancellationToken);
|
||||
|
||||
ValueTask IResettableExecutor.ResetAsync() => this.ResetAsync();
|
||||
}
|
||||
|
||||
internal sealed class WorkflowConcurrentEndExecutor : Executor, IResettableExecutor
|
||||
{
|
||||
public const string ExecutorId = "ConcurrentEnd";
|
||||
|
||||
private readonly int _expectedInputs;
|
||||
private readonly Func<IList<List<ChatMessage>>, List<ChatMessage>> _aggregator;
|
||||
private List<List<ChatMessage>> _allResults;
|
||||
private int _remaining;
|
||||
|
||||
public WorkflowConcurrentEndExecutor(
|
||||
int expectedInputs,
|
||||
Func<IList<List<ChatMessage>>, List<ChatMessage>> aggregator)
|
||||
: base(ExecutorId)
|
||||
{
|
||||
_expectedInputs = expectedInputs;
|
||||
_aggregator = aggregator;
|
||||
_allResults = new List<List<ChatMessage>>(expectedInputs);
|
||||
_remaining = expectedInputs;
|
||||
}
|
||||
|
||||
protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)
|
||||
{
|
||||
protocolBuilder.RouteBuilder.AddHandler<List<ChatMessage>>(async (messages, context, cancellationToken) =>
|
||||
{
|
||||
bool done;
|
||||
lock (_allResults)
|
||||
{
|
||||
_allResults.Add(messages);
|
||||
done = --_remaining == 0;
|
||||
}
|
||||
|
||||
if (!done)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_remaining = _expectedInputs;
|
||||
List<List<ChatMessage>> results = _allResults;
|
||||
_allResults = new List<List<ChatMessage>>(_expectedInputs);
|
||||
await context.YieldOutputAsync(_aggregator(results), cancellationToken).ConfigureAwait(false);
|
||||
});
|
||||
|
||||
return protocolBuilder.YieldsOutput<List<ChatMessage>>();
|
||||
}
|
||||
|
||||
public ValueTask ResetAsync()
|
||||
{
|
||||
_allResults = new List<List<ChatMessage>>(_expectedInputs);
|
||||
_remaining = _expectedInputs;
|
||||
return default;
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class WorkflowRoundRobinGroupChatHost(
|
||||
string id,
|
||||
AIAgent[] agents,
|
||||
Dictionary<AIAgent, ExecutorBinding> agentMap,
|
||||
int maximumIterations)
|
||||
: ChatProtocolExecutor(id, s_options), IResettableExecutor
|
||||
{
|
||||
private static readonly ChatProtocolExecutorOptions s_options = new()
|
||||
{
|
||||
StringMessageChatRole = ChatRole.User,
|
||||
AutoSendTurnToken = false,
|
||||
};
|
||||
|
||||
private readonly AIAgent[] _agents = agents;
|
||||
private readonly Dictionary<AIAgent, ExecutorBinding> _agentMap = agentMap;
|
||||
private readonly int _maximumIterations = maximumIterations;
|
||||
private int _iterationCount;
|
||||
private int _nextIndex;
|
||||
|
||||
protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)
|
||||
=> base.ConfigureProtocol(protocolBuilder).YieldsOutput<List<ChatMessage>>();
|
||||
|
||||
protected override async ValueTask TakeTurnAsync(
|
||||
List<ChatMessage> messages,
|
||||
IWorkflowContext context,
|
||||
bool? emitEvents,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_iterationCount < _maximumIterations)
|
||||
{
|
||||
AIAgent nextAgent = _agents[_nextIndex];
|
||||
_nextIndex = (_nextIndex + 1) % _agents.Length;
|
||||
|
||||
if (_agentMap.TryGetValue(nextAgent, out ExecutorBinding? executor))
|
||||
{
|
||||
_iterationCount++;
|
||||
await context.SendMessageAsync(messages, executor.Id, cancellationToken).ConfigureAwait(false);
|
||||
await context.SendMessageAsync(new TurnToken(emitEvents), executor.Id, cancellationToken).ConfigureAwait(false);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
_iterationCount = 0;
|
||||
_nextIndex = 0;
|
||||
await context.YieldOutputAsync(messages, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
protected override ValueTask ResetAsync()
|
||||
{
|
||||
_iterationCount = 0;
|
||||
_nextIndex = 0;
|
||||
return base.ResetAsync();
|
||||
}
|
||||
|
||||
ValueTask IResettableExecutor.ResetAsync() => this.ResetAsync();
|
||||
}
|
||||
Reference in New Issue
Block a user