diff --git a/sidecar/src/Aryx.AgentHost/Services/CopilotAgentBundle.cs b/sidecar/src/Aryx.AgentHost/Services/CopilotAgentBundle.cs index f0620b8..6c73a7a 100644 --- a/sidecar/src/Aryx.AgentHost/Services/CopilotAgentBundle.cs +++ b/sidecar/src/Aryx.AgentHost/Services/CopilotAgentBundle.cs @@ -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 agents = ResolveOrderedAgents(pattern); + List 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 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> 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 agentMap = agents.ToDictionary( + agent => agent, + CreateAgentExecutorBinding); - return AgentWorkflowBuilder - .CreateGroupChatBuilderWith(agents => - new RoundRobinGroupChatManager(agents) - { - MaximumIterationCount = maximumIterations, - }) - .AddParticipants(ResolveOrderedAgents(pattern).ToArray()) - .Build(); + Func> 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> factory = + (_, __) => new(new WorkflowAggregateTurnMessagesExecutor(id)); + return factory.BindExecutor(id); + } + + private static List AggregateConcurrentResults(IList> lists) + => [.. from list in lists where list.Count > 0 select list.Last()]; + private IReadOnlyList ResolveOrderedAgents(PatternDefinitionDto pattern) { Dictionary agentMap = BuildAgentMap(pattern); diff --git a/sidecar/src/Aryx.AgentHost/Services/WorkflowHostExecutors.cs b/sidecar/src/Aryx.AgentHost/Services/WorkflowHostExecutors.cs new file mode 100644 index 0000000..6220aa6 --- /dev/null +++ b/sidecar/src/Aryx.AgentHost/Services/WorkflowHostExecutors.cs @@ -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>(); + + protected override ValueTask TakeTurnAsync( + List 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 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>, List> _aggregator; + private List> _allResults; + private int _remaining; + + public WorkflowConcurrentEndExecutor( + int expectedInputs, + Func>, List> aggregator) + : base(ExecutorId) + { + _expectedInputs = expectedInputs; + _aggregator = aggregator; + _allResults = new List>(expectedInputs); + _remaining = expectedInputs; + } + + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) + { + protocolBuilder.RouteBuilder.AddHandler>(async (messages, context, cancellationToken) => + { + bool done; + lock (_allResults) + { + _allResults.Add(messages); + done = --_remaining == 0; + } + + if (!done) + { + return; + } + + _remaining = _expectedInputs; + List> results = _allResults; + _allResults = new List>(_expectedInputs); + await context.YieldOutputAsync(_aggregator(results), cancellationToken).ConfigureAwait(false); + }); + + return protocolBuilder.YieldsOutput>(); + } + + public ValueTask ResetAsync() + { + _allResults = new List>(_expectedInputs); + _remaining = _expectedInputs; + return default; + } +} + +internal sealed class WorkflowRoundRobinGroupChatHost( + string id, + AIAgent[] agents, + Dictionary 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 _agentMap = agentMap; + private readonly int _maximumIterations = maximumIterations; + private int _iterationCount; + private int _nextIndex; + + protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) + => base.ConfigureProtocol(protocolBuilder).YieldsOutput>(); + + protected override async ValueTask TakeTurnAsync( + List 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(); +} diff --git a/sidecar/tests/Aryx.AgentHost.Tests/CopilotAgentBundleTests.cs b/sidecar/tests/Aryx.AgentHost.Tests/CopilotAgentBundleTests.cs index a19ae80..41a61b7 100644 --- a/sidecar/tests/Aryx.AgentHost.Tests/CopilotAgentBundleTests.cs +++ b/sidecar/tests/Aryx.AgentHost.Tests/CopilotAgentBundleTests.cs @@ -130,6 +130,36 @@ public sealed class CopilotAgentBundleTests Assert.Equal(HandoffWorkflowGuidance.CreateWorkflowInstructions(), builder.HandoffInstructions); } + [Theory] + [InlineData("single", 1)] + [InlineData("sequential", 2)] + [InlineData("concurrent", 2)] + [InlineData("group-chat", 2)] + public void BuildWorkflow_ExplicitlyConfiguresAgentHostOptions(string mode, int agentCount) + { + CopilotAgentBundle bundle = new(CreateAgents(agentCount), hasConfiguredHooks: false); + PatternDefinitionDto pattern = CreatePattern(mode, agentCount); + + Workflow workflow = bundle.BuildWorkflow(pattern); + + AIAgentBinding[] bindings = workflow.ReflectExecutors().Values + .OfType() + .ToArray(); + + Assert.Equal(agentCount, bindings.Length); + + foreach (AIAgentBinding binding in bindings) + { + AIAgentHostOptions options = Assert.IsType(binding.Options); + Assert.Null(options.EmitAgentUpdateEvents); + Assert.False(options.EmitAgentResponseEvents); + Assert.False(options.InterceptUserInputRequests); + Assert.False(options.InterceptUnterminatedFunctionCalls); + Assert.True(options.ReassignOtherAgentsAsUsers); + Assert.True(options.ForwardIncomingMessages); + } + } + [Fact] public void ConvertToolRequestsToFunctionCalls_MapsCallIdsNamesAndArguments() { @@ -388,6 +418,33 @@ public sealed class CopilotAgentBundleTests CreateTool().JsonSchema); } + private static IReadOnlyList CreateAgents(int count) + => Enumerable.Range(1, count) + .Select(index => (AIAgent)CreateChatClientAgent($"agent-{index}", $"Agent {index}")) + .ToArray(); + + private static PatternDefinitionDto CreatePattern(string mode, int agentCount) + { + return new PatternDefinitionDto + { + Id = $"pattern-{mode}", + Name = $"Pattern {mode}", + Mode = mode, + Availability = "available", + Agents = + [ + .. Enumerable.Range(1, agentCount).Select(index => new PatternAgentDefinitionDto + { + Id = $"agent-{index}", + Name = $"Agent {index}", + Description = $"Agent {index} description.", + Instructions = $"Agent {index} instructions.", + Model = "gpt-5.4", + }), + ], + }; + } + private static ChatClientAgent CreateChatClientAgent(string id, string name) { return new ChatClientAgent(