agent/agent-framework
2
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into feature-harness

Browse Source
This commit is contained in:
westey
2026-04-16 16:46:17 +01:00
committed by GitHub
parent d0ac1d83bc 43d98974d3
commit 99627e41d2
97 changed files with 8769 additions and 638 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dotnet/Directory.Build.targets
+3 -2
View File
@@ -4,8 +4,9 @@
<!-- https://learn.microsoft.com/en-us/nuget/consume-packages/Central-Package-Management -->
<Sdk Name="Microsoft.Build.CentralPackageVersions" Version="2.1.3" />
<!-- Only run 'dotnet format' on dev machines, Release builds. Skip on GitHub Actions -->
<!-- as this runs in its own Actions job. -->
<Target Name="DotnetFormatOnBuild" BeforeTargets="Build" Condition=" '$(Configuration)' == 'Release' AND '$(GITHUB_ACTIONS)' == '' ">
<!-- as this runs in its own Actions job. Only run for net10.0 target frameworks since the dotnet format command -->
<!-- already formats all target frameworks in project. Otherwise it will run format x times x where x is the number of target frameworks -->
<Target Name="DotnetFormatOnBuild" BeforeTargets="Build" Condition=" '$(Configuration)' == 'Release' AND '$(GITHUB_ACTIONS)' == '' AND '$(TargetFramework)' == 'net10.0' ">
<Message Text="Running dotnet format" Importance="high" />
<Exec Command="dotnet format --no-restore -v diag $(ProjectFileName)" />
</Target>
dotnet/Directory.Packages.props
+13 -24
View File
@@ -11,8 +11,8 @@
</PropertyGroup>
<ItemGroup>
<!-- Aspire.* -->
<PackageVersion Include="Anthropic" Version="12.11.0" />
<PackageVersion Include="Anthropic.Foundry" Version="0.4.2" />
<PackageVersion Include="Anthropic" Version="12.13.0" />
<PackageVersion Include="Anthropic.Foundry" Version="0.5.0" />
<PackageVersion Include="Aspire.Azure.AI.OpenAI" Version="13.0.0-preview.1.25560.3" />
<PackageVersion Include="Aspire.Hosting.AppHost" Version="$(AspireAppHostSdkVersion)" />
<PackageVersion Include="Aspire.Hosting.Azure.CognitiveServices" Version="$(AspireAppHostSdkVersion)" />
@@ -32,19 +32,19 @@
<!-- Newtonsoft.Json -->
<PackageVersion Include="Newtonsoft.Json" Version="13.0.4" />
<!-- System.* -->
<PackageVersion Include="Microsoft.Bcl.AsyncInterfaces" Version="10.0.4" />
<PackageVersion Include="Microsoft.Bcl.AsyncInterfaces" Version="10.0.6" />
<PackageVersion Include="Microsoft.Bcl.HashCode" Version="6.0.0" />
<PackageVersion Include="Microsoft.Bcl.Memory" Version="10.0.4" />
<PackageVersion Include="System.ClientModel" Version="1.10.0" />
<PackageVersion Include="System.CodeDom" Version="10.0.0" />
<PackageVersion Include="System.Collections.Immutable" Version="10.0.1" />
<PackageVersion Include="System.CommandLine" Version="2.0.0-rc.2.25502.107" />
<PackageVersion Include="System.Diagnostics.DiagnosticSource" Version="10.0.4" />
<PackageVersion Include="System.Diagnostics.DiagnosticSource" Version="10.0.6" />
<PackageVersion Include="System.Linq.AsyncEnumerable" Version="10.0.4" />
<PackageVersion Include="System.Net.Http.Json" Version="10.0.0" />
<PackageVersion Include="System.Net.ServerSentEvents" Version="10.0.4" />
<PackageVersion Include="System.Text.Json" Version="10.0.4" />
<PackageVersion Include="System.Threading.Channels" Version="10.0.4" />
<PackageVersion Include="System.Text.Json" Version="10.0.6" />
<PackageVersion Include="System.Threading.Channels" Version="10.0.6" />
<PackageVersion Include="System.Threading.Tasks.Extensions" Version="4.6.3" />
<PackageVersion Include="System.Net.Security" Version="4.3.2" />
<!-- OpenTelemetry -->
@@ -63,37 +63,28 @@
<PackageVersion Include="Microsoft.AspNetCore.OpenApi" Version="10.0.0" />
<PackageVersion Include="Swashbuckle.AspNetCore.SwaggerUI" Version="10.0.0" />
<!-- Microsoft.Extensions.* -->
<PackageVersion Include="Microsoft.Extensions.AI" Version="10.4.0" />
<PackageVersion Include="Microsoft.Extensions.AI.Abstractions" Version="10.4.0" />
<PackageVersion Include="Microsoft.Extensions.AI.Evaluation" Version="10.4.0" />
<PackageVersion Include="Microsoft.Extensions.AI.Evaluation.Quality" Version="10.4.0" />
<PackageVersion Include="Microsoft.Extensions.AI.Evaluation.Safety" Version="10.3.0-preview.1.26109.11" />
<PackageVersion Include="Microsoft.Extensions.AI.OpenAI" Version="10.4.0" />
<PackageVersion Include="Microsoft.Extensions.AI" Version="10.5.0" />
<PackageVersion Include="Microsoft.Extensions.AI.Abstractions" Version="10.5.0" />
<PackageVersion Include="Microsoft.Extensions.AI.OpenAI" Version="10.5.0" />
<PackageVersion Include="Microsoft.Extensions.Caching.Memory" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.Compliance.Abstractions" Version="10.4.0" />
<PackageVersion Include="Microsoft.Extensions.Compliance.Abstractions" Version="10.5.0" />
<PackageVersion Include="Microsoft.Extensions.Configuration" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.Configuration.Binder" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.Configuration.EnvironmentVariables" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.Configuration.Json" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.Configuration.UserSecrets" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.4" />
<PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.6" />
<PackageVersion Include="Microsoft.Extensions.Hosting" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.Http.Resilience" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.Logging" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.4" />
<PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.6" />
<PackageVersion Include="Microsoft.Extensions.Logging.Console" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.ServiceDiscovery" Version="10.0.0" />
<PackageVersion Include="Microsoft.Extensions.VectorData.Abstractions" Version="9.7.0" />
<!-- Vector Stores -->
<PackageVersion Include="Microsoft.SemanticKernel.Connectors.InMemory" Version="1.67.0-preview" />
<PackageVersion Include="Microsoft.SemanticKernel.Connectors.Qdrant" Version="1.67.0-preview" />
<!-- Semantic Kernel -->
<PackageVersion Include="Microsoft.SemanticKernel" Version="1.67.0" />
<PackageVersion Include="Microsoft.SemanticKernel.Agents.Core" Version="1.67.0" />
<PackageVersion Include="Microsoft.SemanticKernel.Agents.OpenAI" Version="1.67.0-preview" />
<PackageVersion Include="Microsoft.SemanticKernel.Agents.AzureAI" Version="1.67.0-preview" />
<PackageVersion Include="Microsoft.SemanticKernel.Plugins.OpenApi" Version="1.67.0" />
<!-- Agent SDKs -->
<PackageVersion Include="GitHub.Copilot.SDK" Version="0.1.29" />
<PackageVersion Include="Microsoft.Agents.CopilotStudio.Client" Version="1.3.171-beta" />
@@ -107,11 +98,10 @@
<!-- MCP -->
<PackageVersion Include="ModelContextProtocol" Version="1.1.0" />
<!-- Inference SDKs -->
<PackageVersion Include="AWSSDK.Extensions.Bedrock.MEAI" Version="4.0.5.1" />
<PackageVersion Include="Microsoft.ML.OnnxRuntimeGenAI" Version="0.10.0" />
<PackageVersion Include="Microsoft.ML.Tokenizers" Version="2.0.0" />
<PackageVersion Include="OllamaSharp" Version="5.4.8" />
<PackageVersion Include="OpenAI" Version="2.9.1" />
<PackageVersion Include="OpenAI" Version="2.10.0" />
<!-- Identity -->
<PackageVersion Include="Microsoft.Identity.Client.Extensions.Msal" Version="4.83.1" />
<!-- Workflows -->
@@ -126,7 +116,6 @@
<PackageVersion Include="Microsoft.DurableTask.Worker.AzureManaged" Version="1.18.0" />
<!-- Azure Functions -->
<PackageVersion Include="Microsoft.Azure.Functions.Worker" Version="2.50.0" />
<PackageVersion Include="Microsoft.Azure.Functions.Worker.ApplicationInsights" Version="2.50.0" />
<PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask" Version="1.12.1" />
<PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask.AzureManaged" Version="1.0.1" />
<PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.Http" Version="3.3.0" />
dotnet/samples/05-end-to-end/AGUIClientServer/AGUIServer/Program.cs
+13 -4
View File
@@ -4,6 +4,7 @@ using System.ComponentModel;
using AGUIServer;
using Azure.AI.OpenAI;
using Azure.Identity;
using Microsoft.Agents.AI.Hosting;
using Microsoft.Agents.AI.Hosting.AGUI.AspNetCore;
using Microsoft.Extensions.AI;
using OpenAI.Chat;
@@ -13,11 +14,11 @@ builder.Services.AddHttpClient().AddLogging();
builder.Services.ConfigureHttpJsonOptions(options => options.SerializerOptions.TypeInfoResolverChain.Add(AGUIServerSerializerContext.Default));
builder.Services.AddAGUI();
WebApplication app = builder.Build();
string endpoint = builder.Configuration["AZURE_OPENAI_ENDPOINT"] ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
string deploymentName = builder.Configuration["AZURE_OPENAI_DEPLOYMENT_NAME"] ?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT_NAME is not set.");
const string AgentName = "AGUIAssistant";
// Create the AI agent with tools
// WARNING: DefaultAzureCredential is convenient for development but requires careful consideration in production.
// In production, consider using a specific credential (e.g., ManagedIdentityCredential) to avoid
@@ -27,7 +28,7 @@ var agent = new AzureOpenAIClient(
new DefaultAzureCredential())
.GetChatClient(deploymentName)
.AsAIAgent(
name: "AGUIAssistant",
name: AgentName,
tools: [
AIFunctionFactory.Create(
() => DateTimeOffset.UtcNow,
@@ -48,7 +49,15 @@ var agent = new AzureOpenAIClient(
AGUIServerSerializerContext.Default.Options)
]);
// Register the agent with the host and configure it to use an in-memory session store
// so that conversation state is maintained across requests. In production, you may want to use a persistent session store.
builder
.AddAIAgent(AgentName, (_, _) => agent)
.WithInMemorySessionStore();
WebApplication app = builder.Build();
// Map the AG-UI agent endpoint
app.MapAGUI("/", agent);
app.MapAGUI(AgentName, "/");
await app.RunAsync();
dotnet/src/Microsoft.Agents.AI.Anthropic/Microsoft.Agents.AI.Anthropic.csproj
+1 -1
View File
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<IsReleaseCandidate>true</IsReleaseCandidate>
<IsReleaseCandidate>false</IsReleaseCandidate>
<ImplicitUsings>enable</ImplicitUsings>
<InjectSharedThrow>true</InjectSharedThrow>
</PropertyGroup>
dotnet/src/Microsoft.Agents.AI.Hosting.AGUI.AspNetCore/AGUIEndpointRouteBuilderExtensions.cs
+77 -3
View File
@@ -1,9 +1,12 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Agents.AI.Hosting.AGUI.AspNetCore.Shared;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Http;
@@ -21,6 +24,42 @@ namespace Microsoft.Agents.AI.Hosting.AGUI.AspNetCore;
/// </summary>
public static class AGUIEndpointRouteBuilderExtensions
{
/// <summary>
/// Maps an AG-UI agent endpoint using an agent registered in dependency injection via <see cref="IHostedAgentBuilder"/>.
/// </summary>
/// <param name="endpoints">The endpoint route builder.</param>
/// <param name="agentBuilder">The hosted agent builder that identifies the agent registration.</param>
/// <param name="pattern">The URL pattern for the endpoint.</param>
/// <returns>An <see cref="IEndpointConventionBuilder"/> for the mapped endpoint.</returns>
public static IEndpointConventionBuilder MapAGUI(
this IEndpointRouteBuilder endpoints,
IHostedAgentBuilder agentBuilder,
[StringSyntax("route")] string pattern)
{
ArgumentNullException.ThrowIfNull(endpoints);
ArgumentNullException.ThrowIfNull(agentBuilder);
return endpoints.MapAGUI(agentBuilder.Name, pattern);
}
/// <summary>
/// Maps an AG-UI agent endpoint using a named agent registered in dependency injection.
/// </summary>
/// <param name="endpoints">The endpoint route builder.</param>
/// <param name="agentName">The name of the keyed agent registration to resolve from dependency injection.</param>
/// <param name="pattern">The URL pattern for the endpoint.</param>
/// <returns>An <see cref="IEndpointConventionBuilder"/> for the mapped endpoint.</returns>
public static IEndpointConventionBuilder MapAGUI(
this IEndpointRouteBuilder endpoints,
string agentName,
[StringSyntax("route")] string pattern)
{
ArgumentNullException.ThrowIfNull(endpoints);
ArgumentNullException.ThrowIfNull(agentName);
var agent = endpoints.ServiceProvider.GetRequiredKeyedService<AIAgent>(agentName);
return endpoints.MapAGUI(pattern, agent);
}
/// <summary>
/// Maps an AG-UI agent endpoint.
/// </summary>
@@ -28,11 +67,24 @@ public static class AGUIEndpointRouteBuilderExtensions
/// <param name="pattern">The URL pattern for the endpoint.</param>
/// <param name="aiAgent">The agent instance.</param>
/// <returns>An <see cref="IEndpointConventionBuilder"/> for the mapped endpoint.</returns>
/// <remarks>
/// <para>
/// If an <see cref="AgentSessionStore"/> is registered in dependency injection keyed by the agent's name,
/// it will be used to persist conversation sessions across requests using the AG-UI thread ID as the
/// conversation identifier. If no session store is registered, sessions are ephemeral (not persisted).
/// </para>
/// </remarks>
public static IEndpointConventionBuilder MapAGUI(
this IEndpointRouteBuilder endpoints,
[StringSyntax("route")] string pattern,
AIAgent aiAgent)
{
ArgumentNullException.ThrowIfNull(endpoints);
ArgumentNullException.ThrowIfNull(aiAgent);
var agentSessionStore = endpoints.ServiceProvider.GetKeyedService<AgentSessionStore>(aiAgent.Name);
var hostAgent = new AIHostAgent(aiAgent, agentSessionStore ?? new NoopAgentSessionStore());
return endpoints.MapPost(pattern, async ([FromBody] RunAgentInput? input, HttpContext context, CancellationToken cancellationToken) =>
{
if (input is null)
@@ -63,21 +115,43 @@ public static class AGUIEndpointRouteBuilderExtensions
}
};
var threadId = string.IsNullOrWhiteSpace(input.ThreadId) ? Guid.NewGuid().ToString("N") : input.ThreadId;
var session = await hostAgent.GetOrCreateSessionAsync(threadId, cancellationToken).ConfigureAwait(false);
// Run the agent and convert to AG-UI events
var events = aiAgent.RunStreamingAsync(
var events = hostAgent.RunStreamingAsync(
messages,
session: session,
options: runOptions,
cancellationToken: cancellationToken)
.AsChatResponseUpdatesAsync()
.FilterServerToolsFromMixedToolInvocationsAsync(clientTools, cancellationToken)
.AsAGUIEventStreamAsync(
input.ThreadId,
threadId,
input.RunId,
jsonSerializerOptions,
cancellationToken);
// Wrap the event stream to save the session after streaming completes
var eventsWithSessionSave = SaveSessionAfterStreamingAsync(events, hostAgent, threadId, session, cancellationToken);
var sseLogger = context.RequestServices.GetRequiredService<ILogger<AGUIServerSentEventsResult>>();
return new AGUIServerSentEventsResult(events, sseLogger);
return new AGUIServerSentEventsResult(eventsWithSessionSave, sseLogger);
});
}
private static async IAsyncEnumerable<BaseEvent> SaveSessionAfterStreamingAsync(
IAsyncEnumerable<BaseEvent> events,
AIHostAgent hostAgent,
string threadId,
AgentSession session,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
await foreach (BaseEvent evt in events.ConfigureAwait(false))
{
yield return evt;
}
await hostAgent.SaveSessionAsync(threadId, session, cancellationToken).ConfigureAwait(false);
}
}
dotnet/src/Microsoft.Agents.AI.Hosting.AGUI.AspNetCore/Microsoft.Agents.AI.Hosting.AGUI.AspNetCore.csproj
+1
View File
@@ -19,6 +19,7 @@
<ItemGroup>
<ProjectReference Include="..\Microsoft.Agents.AI\Microsoft.Agents.AI.csproj" />
<ProjectReference Include="..\Microsoft.Agents.AI.Hosting\Microsoft.Agents.AI.Hosting.csproj" />
</ItemGroup>
<ItemGroup>
dotnet/src/Microsoft.Agents.AI.Workflows/HandoffWorkflowBuilder.cs
+60 -17
View File
@@ -8,6 +8,10 @@ using Microsoft.Agents.AI.Workflows.Specialized;
using Microsoft.Extensions.AI;
using Microsoft.Shared.Diagnostics;
using ExecutorFactoryFunc = System.Func<Microsoft.Agents.AI.Workflows.ExecutorConfig<Microsoft.Agents.AI.Workflows.ExecutorOptions>,
string,
System.Threading.Tasks.ValueTask<Microsoft.Agents.AI.Workflows.Specialized.HandoffAgentExecutor>>;
namespace Microsoft.Agents.AI.Workflows;
internal static class DiagnosticConstants
@@ -233,6 +237,57 @@ public class HandoffWorkflowBuilderCore<TBuilder> where TBuilder : HandoffWorkfl
return (TBuilder)this;
}
private Dictionary<string, ExecutorBinding> CreateExecutorBindings(WorkflowBuilder builder)
{
HandoffAgentExecutorOptions options = new(this.HandoffInstructions,
this._emitAgentResponseEvents,
this._emitAgentResponseUpdateEvents,
this._toolCallFilteringBehavior);
// There are two types of ids being used in this method, and it is critical that we are clear about
// which one we are using, and where.
// AgentId...: comes from AIAgent.Id, is often an unreadable machine identifier (e.g. a Guid), and is used to address
// the handoffs
// ExecutorId: uses AIAgent.GetDescriptiveId() to use a friendlier name in telemetry, and is used for ExecutorBinding,
// which are subsequently used in building the workflow
// The outgoing dictionary maps from AgentId => ExecutorBinding
return this._allAgents.ToDictionary(keySelector: a => a.Id, elementSelector: CreateFactoryBinding);
ExecutorBinding CreateFactoryBinding(AIAgent agent)
{
if (!this._targets.TryGetValue(agent, out HashSet<HandoffTarget>? handoffs))
{
handoffs = new();
}
// Use the ExecutorId as the placeholder id for a (possibly) future-bound factory
builder.AddSwitch(HandoffAgentExecutor.IdFor(agent), (SwitchBuilder sb) =>
{
foreach (HandoffTarget handoff in handoffs)
{
sb.AddCase<HandoffState>(state => state?.RequestedHandoffTargetAgentId == handoff.Target.Id, // Use AgentId for target matching
HandoffAgentExecutor.IdFor(handoff.Target)); // Use ExecutorId in for routing at the workflow level
}
sb.WithDefault(HandoffEndExecutor.ExecutorId);
});
ExecutorFactoryFunc factory =
(config, sessionId) => new(
new HandoffAgentExecutor(agent,
handoffs,
options));
// Make sure to use ExecutorId when binding the executor, not AgentId
ExecutorBinding binding = factory.BindExecutor(HandoffAgentExecutor.IdFor(agent));
builder.BindExecutor(binding);
return binding;
}
}
/// <summary>
/// Builds a <see cref="Workflow"/> composed of agents that operate via handoffs, with the next
/// agent to process messages selected by the current agent.
@@ -240,17 +295,12 @@ public class HandoffWorkflowBuilderCore<TBuilder> where TBuilder : HandoffWorkfl
/// <returns>The workflow built based on the handoffs in the builder.</returns>
public Workflow Build()
{
HandoffsStartExecutor start = new(this._returnToPrevious);
HandoffsEndExecutor end = new(this._returnToPrevious);
HandoffStartExecutor start = new(this._returnToPrevious);
HandoffEndExecutor end = new(this._returnToPrevious);
WorkflowBuilder builder = new(start);
HandoffAgentExecutorOptions options = new(this.HandoffInstructions,
this._emitAgentResponseEvents,
this._emitAgentResponseUpdateEvents,
this._toolCallFilteringBehavior);
// Create an AgentExecutor for each agent.
Dictionary<string, HandoffAgentExecutor> executors = this._allAgents.ToDictionary(a => a.Id, a => new HandoffAgentExecutor(a, options));
// Create an factory-based ExecutorBinding for each agent.
Dictionary<string, ExecutorBinding> executors = this.CreateExecutorBindings(builder);
// Connect the start executor to the initial agent (or use dynamic routing when ReturnToPrevious is enabled).
if (this._returnToPrevious)
@@ -263,7 +313,7 @@ public class HandoffWorkflowBuilderCore<TBuilder> where TBuilder : HandoffWorkfl
if (agent.Id != initialAgentId)
{
string agentId = agent.Id;
sb.AddCase<HandoffState>(state => state?.CurrentAgentId == agentId, executors[agentId]);
sb.AddCase<HandoffState>(state => state?.PreviousAgentId == agentId, executors[agentId]);
}
}
@@ -275,13 +325,6 @@ public class HandoffWorkflowBuilderCore<TBuilder> where TBuilder : HandoffWorkfl
builder.AddEdge(start, executors[this._initialAgent.Id]);
}
// Initialize each executor with its handoff targets to the other executors.
foreach (var agent in this._allAgents)
{
executors[agent.Id].Initialize(builder, end, executors,
this._targets.TryGetValue(agent, out HashSet<HandoffTarget>? targets) ? targets : []);
}
// Build the workflow.
return builder.WithOutputFrom(end).Build();
}
dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs
+6
View File
@@ -419,6 +419,12 @@ internal sealed class InProcessRunnerContext : IRunnerContext
.Select(id => this.EnsureExecutorAsync(id, tracer: null).AsTask())
.ToArray();
// Discard queued external deliveries from the superseded timeline so a runtime
// restore cannot apply stale responses after importing the checkpoint state.
while (this._queuedExternalDeliveries.TryDequeue(out _))
{
}
this._nextStep = new StepContext();
this._nextStep.ImportMessages(importedState.QueuedMessages);
dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIAgentHostExecutor.cs
+20 -48
View File
@@ -19,6 +19,9 @@ internal static class TurnExtensions
public static bool ShouldEmitStreamingEvents(bool? turnTokenSetting, bool? agentSetting)
=> turnTokenSetting ?? agentSetting ?? false;
public static bool ShouldEmitStreamingEvents(this HandoffState handoffState, bool? agentSetting)
=> handoffState.TurnToken.ShouldEmitStreamingEvents(agentSetting);
}
internal sealed class AIAgentHostExecutor : ChatProtocolExecutor
@@ -81,7 +84,11 @@ internal sealed class AIAgentHostExecutor : ChatProtocolExecutor
// resumes can be processed in one invocation.
return this.ProcessTurnMessagesAsync(async (pendingMessages, ctx, ct) =>
{
pendingMessages.Add(new ChatMessage(ChatRole.User, [response]));
pendingMessages.Add(new ChatMessage(ChatRole.User, [response])
{
CreatedAt = DateTimeOffset.UtcNow,
MessageId = Guid.NewGuid().ToString("N"),
});
await this.ContinueTurnAsync(pendingMessages, ctx, this._currentTurnEmitEvents ?? false, ct).ConfigureAwait(false);
@@ -104,7 +111,12 @@ internal sealed class AIAgentHostExecutor : ChatProtocolExecutor
// resumes can be processed in one invocation.
return this.ProcessTurnMessagesAsync(async (pendingMessages, ctx, ct) =>
{
pendingMessages.Add(new ChatMessage(ChatRole.Tool, [result]));
pendingMessages.Add(new ChatMessage(ChatRole.Tool, [result])
{
AuthorName = this._agent.Name ?? this._agent.Id,
CreatedAt = DateTimeOffset.UtcNow,
MessageId = Guid.NewGuid().ToString("N"),
});
await this.ContinueTurnAsync(pendingMessages, ctx, this._currentTurnEmitEvents ?? false, ct).ConfigureAwait(false);
@@ -186,16 +198,13 @@ internal sealed class AIAgentHostExecutor : ChatProtocolExecutor
TurnExtensions.ShouldEmitStreamingEvents(turnTokenSetting: emitEvents, this._options.EmitAgentUpdateEvents),
cancellationToken);
private async ValueTask<AgentResponse> InvokeAgentAsync(IEnumerable<ChatMessage> messages, IWorkflowContext context, bool emitEvents, CancellationToken cancellationToken = default)
private async ValueTask<AgentResponse> InvokeAgentAsync(IEnumerable<ChatMessage> messages, IWorkflowContext context, bool emitUpdateEvents, CancellationToken cancellationToken = default)
{
#pragma warning disable MEAI001
Dictionary<string, ToolApprovalRequestContent> userInputRequests = new();
Dictionary<string, FunctionCallContent> functionCalls = new();
AgentResponse response;
AIAgentUnservicedRequestsCollector collector = new(this._userInputHandler, this._functionCallHandler);
if (emitEvents)
if (emitUpdateEvents)
{
#pragma warning disable MEAI001 // Type is for evaluation purposes only and is subject to change or removal in future updates. Suppress this diagnostic to proceed.
// Run the agent in streaming mode only when agent run update events are to be emitted.
IAsyncEnumerable<AgentResponseUpdate> agentStream = this._agent.RunStreamingAsync(
messages,
@@ -206,7 +215,7 @@ internal sealed class AIAgentHostExecutor : ChatProtocolExecutor
await foreach (AgentResponseUpdate update in agentStream.ConfigureAwait(false))
{
await context.YieldOutputAsync(update, cancellationToken).ConfigureAwait(false);
ExtractUnservicedRequests(update.Contents);
collector.ProcessAgentResponseUpdate(update);
updates.Add(update);
}
@@ -220,7 +229,7 @@ internal sealed class AIAgentHostExecutor : ChatProtocolExecutor
cancellationToken: cancellationToken)
.ConfigureAwait(false);
ExtractUnservicedRequests(response.Messages.SelectMany(message => message.Contents));
collector.ProcessAgentResponse(response);
}
if (this._options.EmitAgentResponseEvents)
@@ -228,45 +237,8 @@ internal sealed class AIAgentHostExecutor : ChatProtocolExecutor
await context.YieldOutputAsync(response, cancellationToken).ConfigureAwait(false);
}
if (userInputRequests.Count > 0 || functionCalls.Count > 0)
{
Task userInputTask = this._userInputHandler?.ProcessRequestContentsAsync(userInputRequests, context, cancellationToken) ?? Task.CompletedTask;
Task functionCallTask = this._functionCallHandler?.ProcessRequestContentsAsync(functionCalls, context, cancellationToken) ?? Task.CompletedTask;
await Task.WhenAll(userInputTask, functionCallTask)
.ConfigureAwait(false);
}
await collector.SubmitAsync(context, cancellationToken).ConfigureAwait(false);
return response;
void ExtractUnservicedRequests(IEnumerable<AIContent> contents)
{
foreach (AIContent content in contents)
{
if (content is ToolApprovalRequestContent userInputRequest)
{
// It is an error to simultaneously have multiple outstanding user input requests with the same ID.
userInputRequests.Add(userInputRequest.RequestId, userInputRequest);
}
else if (content is ToolApprovalResponseContent userInputResponse)
{
// If the set of messages somehow already has a corresponding user input response, remove it.
_ = userInputRequests.Remove(userInputResponse.RequestId);
}
else if (content is FunctionCallContent functionCall)
{
// For function calls, we emit an event to notify the workflow.
//
// possibility 1: this will be handled inline by the agent abstraction
// possibility 2: this will not be handled inline by the agent abstraction
functionCalls.Add(functionCall.CallId, functionCall);
}
else if (content is FunctionResultContent functionResult)
{
_ = functionCalls.Remove(functionResult.CallId);
}
}
}
#pragma warning restore MEAI001
}
}
dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIAgentUnservicedRequestsCollector.cs
+78
View File
@@ -0,0 +1,78 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.AI;
namespace Microsoft.Agents.AI.Workflows.Specialized;
internal sealed class AIAgentUnservicedRequestsCollector(AIContentExternalHandler<ToolApprovalRequestContent, ToolApprovalResponseContent>? userInputHandler,
AIContentExternalHandler<FunctionCallContent, FunctionResultContent>? functionCallHandler)
{
private readonly Dictionary<string, ToolApprovalRequestContent> _userInputRequests = [];
private readonly Dictionary<string, FunctionCallContent> _functionCalls = [];
public Task SubmitAsync(IWorkflowContext context, CancellationToken cancellationToken)
{
Task userInputTask = userInputHandler != null && this._userInputRequests.Count > 0
? userInputHandler.ProcessRequestContentsAsync(this._userInputRequests, context, cancellationToken)
: Task.CompletedTask;
Task functionCallTask = functionCallHandler != null && this._functionCalls.Count > 0
? functionCallHandler.ProcessRequestContentsAsync(this._functionCalls, context, cancellationToken)
: Task.CompletedTask;
return Task.WhenAll(userInputTask, functionCallTask);
}
public void ProcessAgentResponseUpdate(AgentResponseUpdate update, Func<FunctionCallContent, bool>? functionCallFilter = null)
=> this.ProcessAIContents(update.Contents, functionCallFilter);
public void ProcessAgentResponse(AgentResponse response)
=> this.ProcessAIContents(response.Messages.SelectMany(message => message.Contents));
public void ProcessAIContents(IEnumerable<AIContent> contents, Func<FunctionCallContent, bool>? functionCallFilter = null)
{
foreach (AIContent content in contents)
{
if (content is ToolApprovalRequestContent userInputRequest)
{
if (this._userInputRequests.ContainsKey(userInputRequest.RequestId))
{
throw new InvalidOperationException($"ToolApprovalRequestContent with duplicate RequestId: {userInputRequest.RequestId}");
}
// It is an error to simultaneously have multiple outstanding user input requests with the same ID.
this._userInputRequests.Add(userInputRequest.RequestId, userInputRequest);
}
else if (content is ToolApprovalResponseContent userInputResponse)
{
// If the set of messages somehow already has a corresponding user input response, remove it.
_ = this._userInputRequests.Remove(userInputResponse.RequestId);
}
else if (content is FunctionCallContent functionCall)
{
// For function calls, we emit an event to notify the workflow.
//
// possibility 1: this will be handled inline by the agent abstraction
// possibility 2: this will not be handled inline by the agent abstraction
if (functionCallFilter == null || functionCallFilter(functionCall))
{
if (this._functionCalls.ContainsKey(functionCall.CallId))
{
throw new InvalidOperationException($"FunctionCallContent with duplicate CallId: {functionCall.CallId}");
}
this._functionCalls.Add(functionCall.CallId, functionCall);
}
}
else if (content is FunctionResultContent functionResult)
{
_ = this._functionCalls.Remove(functionResult.CallId);
}
}
}
}
dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffAgentExecutor.cs
+295 -93
View File
@@ -3,7 +3,6 @@
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Linq;
using System.Text.Json;
@@ -166,128 +165,331 @@ internal sealed class HandoffMessagesFilter
}
}
internal struct AgentInvocationResult(AgentResponse agentResponse, string? handoffTargetId)
{
public AgentResponse Response => agentResponse;
public string? HandoffTargetId => handoffTargetId;
[MemberNotNullWhen(true, nameof(HandoffTargetId))]
public bool IsHandoffRequested => this.HandoffTargetId != null;
}
internal record HandoffAgentHostState(HandoffState? CurrentTurnState, List<ChatMessage> FilteredIncomingMessages, List<ChatMessage> TurnMessages)
{
public HandoffState PrepareHandoff(AgentInvocationResult invocationResult, string currentAgentId)
{
if (this.CurrentTurnState == null)
{
throw new InvalidOperationException("Cannot create a handoff request: Out of turn.");
}
IEnumerable<ChatMessage> allMessages = [.. this.CurrentTurnState.Messages, .. this.TurnMessages, .. invocationResult.Response.Messages];
return new(this.CurrentTurnState.TurnToken, invocationResult.HandoffTargetId, allMessages.ToList(), currentAgentId);
}
}
/// <summary>Executor used to represent an agent in a handoffs workflow, responding to <see cref="HandoffState"/> events.</summary>
[Experimental(DiagnosticConstants.ExperimentalFeatureDiagnostic)]
internal sealed class HandoffAgentExecutor(
AIAgent agent,
HandoffAgentExecutorOptions options) : Executor<HandoffState, HandoffState>(agent.GetDescriptiveId(), declareCrossRunShareable: true), IResettableExecutor
internal sealed class HandoffAgentExecutor :
StatefulExecutor<HandoffAgentHostState, HandoffState>
{
private static readonly JsonElement s_handoffSchema = AIFunctionFactory.Create(
([Description("The reason for the handoff")] string? reasonForHandoff) => { }).JsonSchema;
private readonly AIAgent _agent = agent;
public static string IdFor(AIAgent agent) => agent.GetDescriptiveId();
private readonly AIAgent _agent;
private readonly ChatClientAgentRunOptions? _agentOptions;
private readonly HandoffAgentExecutorOptions _options;
private readonly HashSet<string> _handoffFunctionNames = [];
private readonly Dictionary<string, string> _handoffFunctionToAgentId = [];
private ChatClientAgentRunOptions? _agentOptions;
public void Initialize(
WorkflowBuilder builder,
Executor end,
Dictionary<string, HandoffAgentExecutor> executors,
HashSet<HandoffTarget> handoffs) =>
builder.AddSwitch(this, sb =>
{
if (handoffs.Count != 0)
{
Debug.Assert(this._agentOptions is null);
this._agentOptions = new()
{
ChatOptions = new()
{
AllowMultipleToolCalls = false,
Instructions = options.HandoffInstructions,
Tools = [],
},
};
private static HandoffAgentHostState InitialStateFactory() => new(null, [], []);
int index = 0;
foreach (HandoffTarget handoff in handoffs)
{
index++;
var handoffFunc = AIFunctionFactory.CreateDeclaration($"{HandoffWorkflowBuilder.FunctionPrefix}{index}", handoff.Reason, s_handoffSchema);
this._handoffFunctionNames.Add(handoffFunc.Name);
this._handoffFunctionToAgentId[handoffFunc.Name] = handoff.Target.Id;
this._agentOptions.ChatOptions.Tools.Add(handoffFunc);
sb.AddCase<HandoffState>(state => state?.InvokedHandoff == handoffFunc.Name, executors[handoff.Target.Id]);
}
}
sb.WithDefault(end);
});
public override async ValueTask<HandoffState> HandleAsync(HandoffState message, IWorkflowContext context, CancellationToken cancellationToken = default)
public HandoffAgentExecutor(AIAgent agent, HashSet<HandoffTarget> handoffs, HandoffAgentExecutorOptions options)
: base(IdFor(agent), InitialStateFactory)
{
string? requestedHandoff = null;
List<AgentResponseUpdate> updates = [];
List<ChatMessage> allMessages = message.Messages;
this._agent = agent;
this._options = options;
List<ChatMessage>? roleChanges = allMessages.ChangeAssistantToUserForOtherParticipants(this._agent.Name ?? this._agent.Id);
this._agentOptions = CreateAgentHandoffContext(this._options.HandoffInstructions, handoffs, this._handoffFunctionNames, this._handoffFunctionToAgentId);
}
// If a handoff was invoked by a previous agent, filter out the handoff function
// call and tool result messages before sending to the underlying agent. These
// are internal workflow mechanics that confuse the target model into ignoring the
// original user question.
HandoffMessagesFilter handoffMessagesFilter = new(options.ToolCallFilteringBehavior);
IEnumerable<ChatMessage> messagesForAgent = message.InvokedHandoff is not null
? handoffMessagesFilter.FilterMessages(allMessages)
: allMessages;
private static ChatClientAgentRunOptions? CreateAgentHandoffContext(string? handoffInstructions, HashSet<HandoffTarget> handoffs, HashSet<string> functionNames, Dictionary<string, string> functionToAgentId)
{
ChatClientAgentRunOptions? result = null;
await foreach (var update in this._agent.RunStreamingAsync(messagesForAgent,
options: this._agentOptions,
cancellationToken: cancellationToken)
.ConfigureAwait(false))
if (handoffs.Count != 0)
{
await AddUpdateAsync(update, cancellationToken).ConfigureAwait(false);
foreach (var fcc in update.Contents.OfType<FunctionCallContent>()
.Where(fcc => this._handoffFunctionNames.Contains(fcc.Name)))
result = new()
{
requestedHandoff = fcc.Name;
await AddUpdateAsync(
new AgentResponseUpdate
{
AgentId = this._agent.Id,
AuthorName = this._agent.Name ?? this._agent.Id,
Contents = [new FunctionResultContent(fcc.CallId, "Transferred.")],
CreatedAt = DateTimeOffset.UtcNow,
MessageId = Guid.NewGuid().ToString("N"),
Role = ChatRole.Tool,
},
cancellationToken
)
.ConfigureAwait(false);
ChatOptions = new()
{
AllowMultipleToolCalls = false,
Instructions = handoffInstructions,
Tools = [],
},
};
int index = 0;
foreach (HandoffTarget handoff in handoffs)
{
index++;
var handoffFunc = AIFunctionFactory.CreateDeclaration($"{HandoffWorkflowBuilder.FunctionPrefix}{index}", handoff.Reason, s_handoffSchema);
functionNames.Add(handoffFunc.Name);
functionToAgentId[handoffFunc.Name] = handoff.Target.Id;
result.ChatOptions.Tools.Add(handoffFunc);
}
}
AgentResponse agentResponse = updates.ToAgentResponse();
return result;
}
if (options.EmitAgentResponseEvents)
private AIContentExternalHandler<ToolApprovalRequestContent, ToolApprovalResponseContent>? _userInputHandler;
private AIContentExternalHandler<FunctionCallContent, FunctionResultContent>? _functionCallHandler;
protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)
{
return this.ConfigureUserInputHandling(base.ConfigureProtocol(protocolBuilder))
.SendsMessage<HandoffState>();
}
private ProtocolBuilder ConfigureUserInputHandling(ProtocolBuilder protocolBuilder)
{
this._userInputHandler = new AIContentExternalHandler<ToolApprovalRequestContent, ToolApprovalResponseContent>(
ref protocolBuilder,
portId: $"{this.Id}_UserInput",
intercepted: false,
handler: this.HandleUserInputResponseAsync);
this._functionCallHandler = new AIContentExternalHandler<FunctionCallContent, FunctionResultContent>(
ref protocolBuilder,
portId: $"{this.Id}_FunctionCall",
intercepted: false, // TODO: Use this instead of manual function handling for handoff?
handler: this.HandleFunctionResultAsync);
return protocolBuilder;
}
private ValueTask HandleUserInputResponseAsync(
ToolApprovalResponseContent response,
IWorkflowContext context,
CancellationToken cancellationToken)
{
if (!this._userInputHandler!.MarkRequestAsHandled(response.RequestId))
{
await context.YieldOutputAsync(agentResponse, cancellationToken).ConfigureAwait(false);
throw new InvalidOperationException($"No pending ToolApprovalRequest found with id '{response.RequestId}'.");
}
allMessages.AddRange(agentResponse.Messages);
// Merge the external response with any already-buffered regular messages so mixed-content
// resumes can be processed in one invocation.
return this.InvokeWithStateAsync((state, ctx, ct) =>
{
state.TurnMessages.Add(new ChatMessage(ChatRole.User, [response])
{
CreatedAt = DateTimeOffset.UtcNow,
MessageId = Guid.NewGuid().ToString("N"),
});
return this.ContinueTurnAsync(state, ctx, ct);
}, context, skipCache: false, cancellationToken);
}
private ValueTask HandleFunctionResultAsync(
FunctionResultContent result,
IWorkflowContext context,
CancellationToken cancellationToken)
{
if (!this._functionCallHandler!.MarkRequestAsHandled(result.CallId))
{
throw new InvalidOperationException($"No pending FunctionCall found with id '{result.CallId}'.");
}
// Merge the external response with any already-buffered regular messages so mixed-content
// resumes can be processed in one invocation.
return this.InvokeWithStateAsync((state, ctx, ct) =>
{
state.TurnMessages.Add(
new ChatMessage(ChatRole.Tool, [result])
{
AuthorName = this._agent.Name ?? this._agent.Id,
CreatedAt = DateTimeOffset.UtcNow,
MessageId = Guid.NewGuid().ToString("N"),
});
return this.ContinueTurnAsync(state, ctx, ct);
}, context, skipCache: false, cancellationToken);
}
private async ValueTask<HandoffAgentHostState?> ContinueTurnAsync(HandoffAgentHostState state, IWorkflowContext context, CancellationToken cancellationToken)
{
List<ChatMessage>? roleChanges = state.FilteredIncomingMessages.ChangeAssistantToUserForOtherParticipants(this._agent.Name ?? this._agent.Id);
bool emitUpdateEvents = state.CurrentTurnState!.ShouldEmitStreamingEvents(this._options.EmitAgentResponseUpdateEvents);
AgentInvocationResult result = await this.InvokeAgentAsync([.. state.FilteredIncomingMessages, .. state.TurnMessages], context, emitUpdateEvents, cancellationToken)
.ConfigureAwait(false);
if (this.HasOutstandingRequests && result.IsHandoffRequested)
{
throw new InvalidOperationException("Cannot request a handoff while holding pending requests.");
}
roleChanges.ResetUserToAssistantForChangedRoles();
string currentAgentId = requestedHandoff is not null && this._handoffFunctionToAgentId.TryGetValue(requestedHandoff, out string? targetAgentId)
? targetAgentId
: this._agent.Id;
return new(message.TurnToken, requestedHandoff, allMessages, currentAgentId);
async Task AddUpdateAsync(AgentResponseUpdate update, CancellationToken cancellationToken)
// We send on the HandoffState even if handoff is not requested because we might be terminating the processing, but this only
// happens if we have no outstanding requests.
if (!this.HasOutstandingRequests)
{
updates.Add(update);
if (message.TurnToken.ShouldEmitStreamingEvents(options.EmitAgentResponseUpdateEvents))
HandoffState outgoingState = state.PrepareHandoff(result, this._agent.Id);
await context.SendMessageAsync(outgoingState, cancellationToken).ConfigureAwait(false);
// reset the state for the next handoff (return-to-current is modeled as a new handoff turn, as opposed to "HITL", which
// can be a bit confusing.)
return null;
}
state.TurnMessages.AddRange(result.Response.Messages);
return state;
}
public override ValueTask HandleAsync(HandoffState message, IWorkflowContext context, CancellationToken cancellationToken = default)
{
return this.InvokeWithStateAsync(InvokeContinueTurnAsync, context, skipCache: false, cancellationToken);
ValueTask<HandoffAgentHostState?> InvokeContinueTurnAsync(HandoffAgentHostState state, IWorkflowContext context, CancellationToken cancellationToken)
{
// Check that we are not getting this message while in the middle of a turn
if (state.CurrentTurnState != null)
{
await context.YieldOutputAsync(update, cancellationToken).ConfigureAwait(false);
throw new InvalidOperationException("Cannot have multiple simultaneous conversations in Handoff Orchestration.");
}
// If a handoff was invoked by a previous agent, filter out the handoff function
// call and tool result messages before sending to the underlying agent. These
// are internal workflow mechanics that confuse the target model into ignoring the
// original user question.
HandoffMessagesFilter handoffMessagesFilter = new(this._options.ToolCallFilteringBehavior);
IEnumerable<ChatMessage> messagesForAgent = message.RequestedHandoffTargetAgentId is not null
? handoffMessagesFilter.FilterMessages(message.Messages)
: message.Messages;
// This works because the runtime guarantees that a given executor instance will process messages serially,
// though there is no global cross-executor ordering guarantee (and in turn, no canonical message delivery order)
state = new(message, messagesForAgent.ToList(), []);
return this.ContinueTurnAsync(state, context, cancellationToken);
}
}
public ValueTask ResetAsync() => default;
private const string UserInputRequestStateKey = nameof(_userInputHandler);
private const string FunctionCallRequestStateKey = nameof(_functionCallHandler);
protected internal override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
{
Task userInputRequestsTask = this._userInputHandler?.OnCheckpointingAsync(UserInputRequestStateKey, context, cancellationToken).AsTask() ?? Task.CompletedTask;
Task functionCallRequestsTask = this._functionCallHandler?.OnCheckpointingAsync(FunctionCallRequestStateKey, context, cancellationToken).AsTask() ?? Task.CompletedTask;
Task baseTask = base.OnCheckpointingAsync(context, cancellationToken).AsTask();
await Task.WhenAll(userInputRequestsTask, functionCallRequestsTask, baseTask).ConfigureAwait(false);
}
protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
{
Task userInputRestoreTask = this._userInputHandler?.OnCheckpointRestoredAsync(UserInputRequestStateKey, context, cancellationToken).AsTask() ?? Task.CompletedTask;
Task functionCallRestoreTask = this._functionCallHandler?.OnCheckpointRestoredAsync(FunctionCallRequestStateKey, context, cancellationToken).AsTask() ?? Task.CompletedTask;
await Task.WhenAll(userInputRestoreTask, functionCallRestoreTask).ConfigureAwait(false);
await base.OnCheckpointRestoredAsync(context, cancellationToken).ConfigureAwait(false);
}
private bool HasOutstandingRequests => (this._userInputHandler?.HasPendingRequests == true)
|| (this._functionCallHandler?.HasPendingRequests == true);
private async ValueTask<AgentInvocationResult> InvokeAgentAsync(IEnumerable<ChatMessage> messages, IWorkflowContext context, bool emitUpdateEvents, CancellationToken cancellationToken = default)
{
AgentResponse response;
AIAgentUnservicedRequestsCollector collector = new(this._userInputHandler, this._functionCallHandler);
IAsyncEnumerable<AgentResponseUpdate> agentStream = this._agent.RunStreamingAsync(
messages,
options: this._agentOptions,
cancellationToken: cancellationToken);
string? requestedHandoff = null;
List<AgentResponseUpdate> updates = [];
List<FunctionCallContent> candidateRequests = [];
await foreach (AgentResponseUpdate update in agentStream.ConfigureAwait(false))
{
await AddUpdateAsync(update, cancellationToken).ConfigureAwait(false);
collector.ProcessAgentResponseUpdate(update, CollectHandoffRequestsFilter);
bool CollectHandoffRequestsFilter(FunctionCallContent candidateHandoffRequest)
{
bool isHandoffRequest = this._handoffFunctionNames.Contains(candidateHandoffRequest.Name);
if (isHandoffRequest)
{
candidateRequests.Add(candidateHandoffRequest);
}
return !isHandoffRequest;
}
}
if (candidateRequests.Count > 1)
{
string message = $"Duplicate handoff requests in single turn ([{string.Join(", ", candidateRequests.Select(request => request.Name))}]). Using last ({candidateRequests.Last().Name})";
await context.AddEventAsync(new WorkflowWarningEvent(message), cancellationToken).ConfigureAwait(false);
}
if (candidateRequests.Count > 0)
{
FunctionCallContent handoffRequest = candidateRequests[candidateRequests.Count - 1];
requestedHandoff = handoffRequest.Name;
await AddUpdateAsync(
new AgentResponseUpdate
{
AgentId = this._agent.Id,
AuthorName = this._agent.Name ?? this._agent.Id,
Contents = [new FunctionResultContent(handoffRequest.CallId, "Transferred.")],
CreatedAt = DateTimeOffset.UtcNow,
MessageId = Guid.NewGuid().ToString("N"),
Role = ChatRole.Tool,
},
cancellationToken
)
.ConfigureAwait(false);
}
response = updates.ToAgentResponse();
if (this._options.EmitAgentResponseEvents)
{
await context.YieldOutputAsync(response, cancellationToken).ConfigureAwait(false);
}
await collector.SubmitAsync(context, cancellationToken).ConfigureAwait(false);
return new(response, LookupHandoffTarget(requestedHandoff));
ValueTask AddUpdateAsync(AgentResponseUpdate update, CancellationToken cancellationToken)
{
updates.Add(update);
return emitUpdateEvents ? context.YieldOutputAsync(update, cancellationToken) : default;
}
string? LookupHandoffTarget(string? requestedHandoff)
=> requestedHandoff != null
? this._handoffFunctionToAgentId.TryGetValue(requestedHandoff, out string? targetId) ? targetId : null
: null;
}
}
dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffsEndExecutor.cs → dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffEndExecutor.cs
+4 -4
View File
@@ -8,7 +8,7 @@ using Microsoft.Extensions.AI;
namespace Microsoft.Agents.AI.Workflows.Specialized;
/// <summary>Executor used at the end of a handoff workflow to raise a final completed event.</summary>
internal sealed class HandoffsEndExecutor(bool returnToPrevious) : Executor(ExecutorId, declareCrossRunShareable: true), IResettableExecutor
internal sealed class HandoffEndExecutor(bool returnToPrevious) : Executor(ExecutorId, declareCrossRunShareable: true), IResettableExecutor
{
public const string ExecutorId = "HandoffEnd";
@@ -21,9 +21,9 @@ internal sealed class HandoffsEndExecutor(bool returnToPrevious) : Executor(Exec
{
if (returnToPrevious)
{
await context.QueueStateUpdateAsync<string?>(HandoffConstants.CurrentAgentTrackerKey,
handoff.CurrentAgentId,
HandoffConstants.CurrentAgentTrackerScope,
await context.QueueStateUpdateAsync<string?>(HandoffConstants.PreviousAgentTrackerKey,
handoff.PreviousAgentId,
HandoffConstants.PreviousAgentTrackerScope,
cancellationToken)
.ConfigureAwait(false);
}
dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffsStartExecutor.cs → dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffStartExecutor.cs
+8 -8
View File
@@ -9,12 +9,12 @@ namespace Microsoft.Agents.AI.Workflows.Specialized;
internal static class HandoffConstants
{
internal const string CurrentAgentTrackerKey = "LastAgentId";
internal const string CurrentAgentTrackerScope = "HandoffOrchestration";
internal const string PreviousAgentTrackerKey = "LastAgentId";
internal const string PreviousAgentTrackerScope = "HandoffOrchestration";
}
/// <summary>Executor used at the start of a handoffs workflow to accumulate messages and emit them as HandoffState upon receiving a turn token.</summary>
internal sealed class HandoffsStartExecutor(bool returnToPrevious) : ChatProtocolExecutor(ExecutorId, DefaultOptions, declareCrossRunShareable: true), IResettableExecutor
internal sealed class HandoffStartExecutor(bool returnToPrevious) : ChatProtocolExecutor(ExecutorId, DefaultOptions, declareCrossRunShareable: true), IResettableExecutor
{
internal const string ExecutorId = "HandoffStart";
@@ -32,15 +32,15 @@ internal sealed class HandoffsStartExecutor(bool returnToPrevious) : ChatProtoco
if (returnToPrevious)
{
return context.InvokeWithStateAsync(
async (string? currentAgentId, IWorkflowContext context, CancellationToken cancellationToken) =>
async (string? previousAgentId, IWorkflowContext context, CancellationToken cancellationToken) =>
{
HandoffState handoffState = new(new(emitEvents), null, messages, currentAgentId);
HandoffState handoffState = new(new(emitEvents), null, messages, previousAgentId);
await context.SendMessageAsync(handoffState, cancellationToken).ConfigureAwait(false);
return currentAgentId;
return previousAgentId;
},
HandoffConstants.CurrentAgentTrackerKey,
HandoffConstants.CurrentAgentTrackerScope,
HandoffConstants.PreviousAgentTrackerKey,
HandoffConstants.PreviousAgentTrackerScope,
cancellationToken);
}
dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffState.cs
+2 -2
View File
@@ -7,6 +7,6 @@ namespace Microsoft.Agents.AI.Workflows.Specialized;
internal sealed record class HandoffState(
TurnToken TurnToken,
string? InvokedHandoff,
string? RequestedHandoffTargetAgentId,
List<ChatMessage> Messages,
string? CurrentAgentId = null);
string? PreviousAgentId = null);
dotnet/src/Microsoft.Agents.AI.Workflows/StatefulExecutor.cs
+17 -15
View File
@@ -113,6 +113,12 @@ public abstract class StatefulExecutor<TState> : Executor
{
if (!skipCache && !context.ConcurrentRunsEnabled)
{
if (this._stateCache is null)
{
this._stateCache = await context.ReadOrInitStateAsync(this.StateKey, this._initialStateFactory, this.Options.ScopeName, cancellationToken)
.ConfigureAwait(false);
}
TState newState = await invocation(this._stateCache ?? this._initialStateFactory(),
context,
cancellationToken).ConfigureAwait(false)
@@ -168,9 +174,12 @@ public abstract class StatefulExecutor<TState, TInput>(string id,
/// <inheritdoc/>
protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)
{
protocolBuilder.RouteBuilder.AddHandler<TInput>(this.HandleAsync);
Func<TInput, IWorkflowContext, CancellationToken, ValueTask> handlerDelegate = this.HandleAsync;
return protocolBuilder.SendsMessageTypes(sentMessageTypes ?? [])
return protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler(handlerDelegate))
.AddMethodAttributeTypes(handlerDelegate.Method)
.AddClassAttributeTypes(this.GetType())
.SendsMessageTypes(sentMessageTypes ?? [])
.YieldsOutputTypes(outputTypes ?? []);
}
@@ -203,19 +212,12 @@ public abstract class StatefulExecutor<TState, TInput, TOutput>(string id,
/// <inheritdoc/>
protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)
{
protocolBuilder.RouteBuilder.AddHandler<TInput, TOutput>(this.HandleAsync);
if (this.Options.AutoSendMessageHandlerResultObject)
{
protocolBuilder.SendsMessage<TOutput>();
}
if (this.Options.AutoYieldOutputHandlerResultObject)
{
protocolBuilder.YieldsOutput<TOutput>();
}
return protocolBuilder.SendsMessageTypes(sentMessageTypes ?? []).YieldsOutputTypes(outputTypes ?? []);
Func<TInput, IWorkflowContext, CancellationToken, ValueTask<TOutput>> handlerDelegate = this.HandleAsync;
return protocolBuilder.ConfigureRoutes(routeBuilder => routeBuilder.AddHandler(handlerDelegate))
.AddMethodAttributeTypes(handlerDelegate.Method)
.AddClassAttributeTypes(this.GetType())
.SendsMessageTypes(sentMessageTypes ?? [])
.YieldsOutputTypes(outputTypes ?? []);
}
/// <inheritdoc/>
dotnet/tests/Microsoft.Agents.AI.Anthropic.UnitTests/Extensions/AnthropicBetaServiceExtensionsTests.cs
+8
View File
@@ -483,6 +483,14 @@ public sealed class AnthropicBetaServiceExtensionsTests
public IBetaMessageService Messages => new Mock<IBetaMessageService>().Object;
public global::Anthropic.Services.Beta.IAgentService Agents => throw new NotImplementedException();
public global::Anthropic.Services.Beta.IEnvironmentService Environments => throw new NotImplementedException();
public global::Anthropic.Services.Beta.ISessionService Sessions => throw new NotImplementedException();
public global::Anthropic.Services.Beta.IVaultService Vaults => throw new NotImplementedException();
public IBetaService WithOptions(Func<ClientOptions, ClientOptions> modifier)
{
throw new NotImplementedException();
dotnet/tests/Microsoft.Agents.AI.Hosting.AGUI.AspNetCore.IntegrationTests/Microsoft.Agents.AI.Hosting.AGUI.AspNetCore.IntegrationTests.csproj
+1
View File
@@ -21,6 +21,7 @@
<ItemGroup>
<ProjectReference Include="..\..\src\Microsoft.Agents.AI\Microsoft.Agents.AI.csproj" />
<ProjectReference Include="..\..\src\Microsoft.Agents.AI.Hosting\Microsoft.Agents.AI.Hosting.csproj" />
<ProjectReference Include="..\..\src\Microsoft.Agents.AI.Hosting.AGUI.AspNetCore\Microsoft.Agents.AI.Hosting.AGUI.AspNetCore.csproj" />
<ProjectReference Include="..\..\src\Microsoft.Agents.AI.AGUI\Microsoft.Agents.AI.AGUI.csproj" />
<ProjectReference Include="..\..\src\Microsoft.Agents.AI.OpenAI\Microsoft.Agents.AI.OpenAI.csproj" />
dotnet/tests/Microsoft.Agents.AI.Hosting.AGUI.AspNetCore.IntegrationTests/SessionPersistenceTests.cs
+226
View File
@@ -0,0 +1,226 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Net.Http;
using System.Runtime.CompilerServices;
using System.Text.Json;
using System.Text.Json.Serialization;
using System.Threading;
using System.Threading.Tasks;
using FluentAssertions;
using Microsoft.Agents.AI.AGUI;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting.Server;
using Microsoft.AspNetCore.TestHost;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.DependencyInjection;
namespace Microsoft.Agents.AI.Hosting.AGUI.AspNetCore.IntegrationTests;
public sealed class SessionPersistenceTests : IAsyncDisposable
{
private WebApplication? _app;
private HttpClient? _client;
[Fact]
public async Task MultiTurnWithSessionStore_PersistsSessionAcrossRequestsAsync()
{
// Arrange - use hosting DI pattern with InMemorySessionStore.
// FakeSessionAgent tracks turn count in session StateBag so we can verify
// that state survives the serialization round-trip through the session store.
await this.SetupTestServerWithSessionStoreAsync();
var chatClient = new AGUIChatClient(this._client!, "", null);
AIAgent agent = chatClient.AsAIAgent(instructions: null, name: "assistant", description: "Sample assistant", tools: []);
ChatClientAgentSession session = (ChatClientAgentSession)await agent.CreateSessionAsync();
// Act - First turn
ChatMessage firstUserMessage = new(ChatRole.User, "First message");
List<AgentResponseUpdate> firstTurnUpdates = [];
await foreach (AgentResponseUpdate update in agent.RunStreamingAsync([firstUserMessage], session, new AgentRunOptions(), CancellationToken.None))
{
firstTurnUpdates.Add(update);
}
// Act - Second turn (same thread ID to test session persistence)
ChatMessage secondUserMessage = new(ChatRole.User, "Second message");
List<AgentResponseUpdate> secondTurnUpdates = [];
await foreach (AgentResponseUpdate update in agent.RunStreamingAsync([secondUserMessage], session, new AgentRunOptions(), CancellationToken.None))
{
secondTurnUpdates.Add(update);
}
// Assert - Verify turn count proves session state was persisted.
// If session persistence were broken, both turns would return "Turn 1"
// because a fresh session (with turn count 0) would be created each time.
AgentResponse firstResponse = firstTurnUpdates.ToAgentResponse();
firstResponse.Messages.Should().HaveCount(1);
firstResponse.Messages[0].Role.Should().Be(ChatRole.Assistant);
firstResponse.Messages[0].Text.Should().Contain("Turn 1:");
AgentResponse secondResponse = secondTurnUpdates.ToAgentResponse();
secondResponse.Messages.Should().HaveCount(1);
secondResponse.Messages[0].Role.Should().Be(ChatRole.Assistant);
secondResponse.Messages[0].Text.Should().Contain("Turn 2:");
}
[Fact]
public async Task MapAGUI_WithAgentName_StreamsResponseCorrectlyAsync()
{
// Arrange - use the MapAGUI(agentName, pattern) overload via hosting DI
await this.SetupTestServerWithSessionStoreAsync();
var chatClient = new AGUIChatClient(this._client!, "", null);
AIAgent agent = chatClient.AsAIAgent(instructions: null, name: "assistant", description: "Sample assistant", tools: []);
ChatClientAgentSession session = (ChatClientAgentSession)await agent.CreateSessionAsync();
ChatMessage userMessage = new(ChatRole.User, "hello");
List<AgentResponseUpdate> updates = [];
// Act
await foreach (AgentResponseUpdate update in agent.RunStreamingAsync([userMessage], session, new AgentRunOptions(), CancellationToken.None))
{
updates.Add(update);
}
// Assert
updates.Should().NotBeEmpty();
updates.Should().AllSatisfy(u => u.Role.Should().Be(ChatRole.Assistant));
AgentResponse response = updates.ToAgentResponse();
response.Messages.Should().HaveCount(1);
response.Messages[0].Role.Should().Be(ChatRole.Assistant);
response.Messages[0].Text.Should().Be("Turn 1: Hello from session agent!");
}
private async Task SetupTestServerWithSessionStoreAsync()
{
WebApplicationBuilder builder = WebApplication.CreateBuilder();
builder.WebHost.UseTestServer();
builder.Services.AddAGUI();
// Register agent using hosting DI pattern with InMemorySessionStore
builder.Services.AddAIAgent("session-test-agent", (_, name) => new FakeSessionAgent(name))
.WithInMemorySessionStore();
this._app = builder.Build();
// Use the agentName overload of MapAGUI
this._app.MapAGUI("session-test-agent", "/agent");
await this._app.StartAsync();
TestServer testServer = this._app.Services.GetRequiredService<IServer>() as TestServer
?? throw new InvalidOperationException("TestServer not found");
this._client = testServer.CreateClient();
this._client.BaseAddress = new Uri("http://localhost/agent");
}
public async ValueTask DisposeAsync()
{
this._client?.Dispose();
if (this._app != null)
{
await this._app.DisposeAsync();
}
}
}
[SuppressMessage("Performance", "CA1812:Avoid uninstantiated internal classes", Justification = "Instantiated via dependency injection")]
internal sealed class FakeSessionAgent : AIAgent
{
private readonly string _name;
public FakeSessionAgent(string name)
{
this._name = name;
}
protected override string? IdCore => this._name;
public override string? Name => this._name;
public override string? Description => "A fake agent with session support for testing";
protected override ValueTask<AgentSession> CreateSessionCoreAsync(CancellationToken cancellationToken = default) =>
new(new FakeSessionAgentSession());
protected override ValueTask<AgentSession> DeserializeSessionCoreAsync(JsonElement serializedState, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default) =>
new(serializedState.Deserialize<FakeSessionAgentSession>(jsonSerializerOptions)!);
protected override ValueTask<JsonElement> SerializeSessionCoreAsync(AgentSession session, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default)
{
if (session is not FakeSessionAgentSession fakeSession)
{
throw new InvalidOperationException($"The provided session type '{session.GetType().Name}' is not compatible with this agent.");
}
return new(JsonSerializer.SerializeToElement(fakeSession, jsonSerializerOptions));
}
protected override async Task<AgentResponse> RunCoreAsync(
IEnumerable<ChatMessage> messages,
AgentSession? session = null,
AgentRunOptions? options = null,
CancellationToken cancellationToken = default)
{
List<AgentResponseUpdate> updates = [];
await foreach (AgentResponseUpdate update in this.RunStreamingAsync(messages, session, options, cancellationToken).ConfigureAwait(false))
{
updates.Add(update);
}
return updates.ToAgentResponse();
}
protected override async IAsyncEnumerable<AgentResponseUpdate> RunCoreStreamingAsync(
IEnumerable<ChatMessage> messages,
AgentSession? session = null,
AgentRunOptions? options = null,
[EnumeratorCancellation] CancellationToken cancellationToken = default)
{
// Track turn count in session state to enable persistence verification.
// If the session store works correctly, the turn count increments across requests.
int turnCount = 1;
if (session != null)
{
var counter = session.StateBag.GetValue<TurnCounter>("turnCounter");
turnCount = (counter?.Count ?? 0) + 1;
session.StateBag.SetValue("turnCounter", new TurnCounter { Count = turnCount });
}
string messageId = Guid.NewGuid().ToString("N");
string prefix = $"Turn {turnCount}: ";
foreach (string chunk in new[] { prefix, "Hello", " ", "from", " ", "session", " ", "agent", "!" })
{
yield return new AgentResponseUpdate
{
MessageId = messageId,
Role = ChatRole.Assistant,
Contents = [new TextContent(chunk)]
};
await Task.Yield();
}
}
internal sealed class TurnCounter
{
public int Count { get; set; }
}
private sealed class FakeSessionAgentSession : AgentSession
{
public FakeSessionAgentSession()
{
}
[JsonConstructor]
public FakeSessionAgentSession(AgentSessionStateBag stateBag) : base(stateBag)
{
}
}
}
dotnet/tests/Microsoft.Agents.AI.Hosting.AGUI.AspNetCore.UnitTests/AGUIEndpointRouteBuilderExtensionsTests.cs
+191
View File
@@ -14,6 +14,7 @@ using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Routing;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Moq;
@@ -31,6 +32,7 @@ public sealed class AGUIEndpointRouteBuilderExtensionsTests
// Arrange
Mock<IEndpointRouteBuilder> endpointsMock = new();
Mock<IServiceProvider> serviceProviderMock = new();
serviceProviderMock.As<IKeyedServiceProvider>();
endpointsMock.Setup(e => e.ServiceProvider).Returns(serviceProviderMock.Object);
endpointsMock.Setup(e => e.DataSources).Returns([]);
@@ -45,6 +47,155 @@ public sealed class AGUIEndpointRouteBuilderExtensionsTests
Assert.NotNull(result);
}
[Fact]
public void MapAGUI_WithAgentName_ResolvesKeyedAgentFromDI()
{
// Arrange
Mock<IEndpointRouteBuilder> endpointsMock = new();
Mock<IServiceProvider> serviceProviderMock = new();
AIAgent agent = new NamedTestAgent();
serviceProviderMock.As<IKeyedServiceProvider>()
.Setup(sp => sp.GetRequiredKeyedService(typeof(AIAgent), "test-agent"))
.Returns(agent);
endpointsMock.Setup(e => e.ServiceProvider).Returns(serviceProviderMock.Object);
endpointsMock.Setup(e => e.DataSources).Returns([]);
// Act
IEndpointConventionBuilder? result = endpointsMock.Object.MapAGUI("test-agent", "/api/agent");
// Assert
Assert.NotNull(result);
serviceProviderMock.As<IKeyedServiceProvider>()
.Verify(sp => sp.GetRequiredKeyedService(typeof(AIAgent), "test-agent"), Times.Once);
}
[Fact]
public void MapAGUI_WithHostedAgentBuilder_ResolvesAgentByBuilderName()
{
// Arrange
Mock<IEndpointRouteBuilder> endpointsMock = new();
Mock<IServiceProvider> serviceProviderMock = new();
Mock<IHostedAgentBuilder> agentBuilderMock = new();
AIAgent agent = new NamedTestAgent();
agentBuilderMock.Setup(b => b.Name).Returns("test-agent");
serviceProviderMock.As<IKeyedServiceProvider>()
.Setup(sp => sp.GetRequiredKeyedService(typeof(AIAgent), "test-agent"))
.Returns(agent);
endpointsMock.Setup(e => e.ServiceProvider).Returns(serviceProviderMock.Object);
endpointsMock.Setup(e => e.DataSources).Returns([]);
// Act
IEndpointConventionBuilder? result = endpointsMock.Object.MapAGUI(agentBuilderMock.Object, "/api/agent");
// Assert
Assert.NotNull(result);
serviceProviderMock.As<IKeyedServiceProvider>()
.Verify(sp => sp.GetRequiredKeyedService(typeof(AIAgent), "test-agent"), Times.Once);
}
[Fact]
public void MapAGUI_WithAgent_ResolvesSessionStoreFromDI()
{
// Arrange
Mock<IEndpointRouteBuilder> endpointsMock = new();
Mock<IServiceProvider> serviceProviderMock = new();
Mock<AgentSessionStore> sessionStoreMock = new();
AIAgent agent = new NamedTestAgent();
serviceProviderMock.As<IKeyedServiceProvider>()
.Setup(sp => sp.GetKeyedService(typeof(AgentSessionStore), "test-agent"))
.Returns(sessionStoreMock.Object);
endpointsMock.Setup(e => e.ServiceProvider).Returns(serviceProviderMock.Object);
endpointsMock.Setup(e => e.DataSources).Returns([]);
// Act
IEndpointConventionBuilder? result = endpointsMock.Object.MapAGUI("/api/agent", agent);
// Assert
Assert.NotNull(result);
serviceProviderMock.As<IKeyedServiceProvider>()
.Verify(sp => sp.GetKeyedService(typeof(AgentSessionStore), "test-agent"), Times.Once);
}
[Fact]
public void MapAGUI_WithoutSessionStore_FallsBackToNoopStore()
{
// Arrange
Mock<IEndpointRouteBuilder> endpointsMock = new();
Mock<IServiceProvider> serviceProviderMock = new();
AIAgent agent = new TestAgent();
// No session store registered - IKeyedServiceProvider returns null by default
serviceProviderMock.As<IKeyedServiceProvider>();
endpointsMock.Setup(e => e.ServiceProvider).Returns(serviceProviderMock.Object);
endpointsMock.Setup(e => e.DataSources).Returns([]);
// Act - should not throw (falls back to NoopAgentSessionStore)
IEndpointConventionBuilder? result = endpointsMock.Object.MapAGUI("/api/agent", agent);
// Assert
Assert.NotNull(result);
}
[Fact]
public void MapAGUI_WithNullEndpoints_ThrowsArgumentNullException()
{
// Arrange
AIAgent agent = new TestAgent();
// Act & Assert
Assert.Throws<ArgumentNullException>(() =>
AGUIEndpointRouteBuilderExtensions.MapAGUI(null!, "/api/agent", agent));
}
[Fact]
public void MapAGUI_WithNullAgent_ThrowsArgumentNullException()
{
// Arrange
Mock<IEndpointRouteBuilder> endpointsMock = new();
Mock<IServiceProvider> serviceProviderMock = new();
serviceProviderMock.As<IKeyedServiceProvider>();
endpointsMock.Setup(e => e.ServiceProvider).Returns(serviceProviderMock.Object);
// Act & Assert
Assert.Throws<ArgumentNullException>(() =>
endpointsMock.Object.MapAGUI("/api/agent", (AIAgent)null!));
}
[Fact]
public void MapAGUI_WithNullAgentName_ThrowsArgumentNullException()
{
// Arrange
Mock<IEndpointRouteBuilder> endpointsMock = new();
Mock<IServiceProvider> serviceProviderMock = new();
serviceProviderMock.As<IKeyedServiceProvider>();
endpointsMock.Setup(e => e.ServiceProvider).Returns(serviceProviderMock.Object);
// Act & Assert
Assert.Throws<ArgumentNullException>(() =>
endpointsMock.Object.MapAGUI((string)null!, "/api/agent"));
}
[Fact]
public void MapAGUI_WithNullAgentBuilder_ThrowsArgumentNullException()
{
// Arrange
Mock<IEndpointRouteBuilder> endpointsMock = new();
Mock<IServiceProvider> serviceProviderMock = new();
endpointsMock.Setup(e => e.ServiceProvider).Returns(serviceProviderMock.Object);
// Act & Assert
Assert.Throws<ArgumentNullException>(() =>
endpointsMock.Object.MapAGUI((IHostedAgentBuilder)null!, "/api/agent"));
}
[Fact]
public async Task MapAGUIAgent_WithNullOrInvalidInput_Returns400BadRequestAsync()
{
@@ -556,4 +707,44 @@ public sealed class AGUIEndpointRouteBuilderExtensionsTests
yield return new AgentResponseUpdate(new ChatResponseUpdate(ChatRole.Assistant, "Test response"));
}
}
private sealed class NamedTestAgent : AIAgent
{
protected override string? IdCore => "test-agent";
public override string? Name => "test-agent";
public override string? Description => "Named test agent";
protected override ValueTask<AgentSession> CreateSessionCoreAsync(CancellationToken cancellationToken = default) =>
new(new TestAgentSession());
protected override ValueTask<AgentSession> DeserializeSessionCoreAsync(JsonElement serializedState, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default) =>
new(serializedState.Deserialize<TestAgentSession>(jsonSerializerOptions)!);
protected override ValueTask<JsonElement> SerializeSessionCoreAsync(AgentSession session, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default)
{
if (session is not TestAgentSession testSession)
{
throw new InvalidOperationException($"The provided session type '{session.GetType().Name}' is not compatible with this agent. Only sessions of type '{nameof(TestAgentSession)}' can be serialized by this agent.");
}
return new(JsonSerializer.SerializeToElement(testSession, jsonSerializerOptions));
}
protected override Task<AgentResponse> RunCoreAsync(IEnumerable<ChatMessage> messages, AgentSession? session = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
{
throw new NotImplementedException();
}
protected override async IAsyncEnumerable<AgentResponseUpdate> RunCoreStreamingAsync(
IEnumerable<ChatMessage> messages,
AgentSession? session = null,
AgentRunOptions? options = null,
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
{
await Task.CompletedTask;
yield return new AgentResponseUpdate(new ChatResponseUpdate(ChatRole.Assistant, "Test response"));
}
}
}
dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/AgentWorkflowBuilderTests.cs
+272 -12
View File
@@ -9,6 +9,7 @@ using System.Text.Json;
using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;
using FluentAssertions;
using Microsoft.Agents.AI.Workflows.InProc;
using Microsoft.Extensions.AI;
@@ -147,7 +148,7 @@ public class AgentWorkflowBuilderTests
for (int iter = 0; iter < 3; iter++)
{
const string UserInput = "abc";
(string updateText, List<ChatMessage>? result, _) = await RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, UserInput)]);
(string updateText, List<ChatMessage>? result, _, _) = await RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, UserInput)]);
Assert.NotNull(result);
Assert.Equal(numAgents + 1, result.Count);
@@ -225,7 +226,7 @@ public class AgentWorkflowBuilderTests
barrier.Value = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
remaining.Value = 2;
(string updateText, List<ChatMessage>? result, _) = await RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, "abc")]);
(string updateText, List<ChatMessage>? result, _, _) = await RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, "abc")]);
Assert.NotEmpty(updateText);
Assert.NotNull(result);
@@ -258,7 +259,7 @@ public class AgentWorkflowBuilderTests
}), description: "nop"))
.Build();
(string updateText, List<ChatMessage>? result, _) = await RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, "abc")]);
(string updateText, List<ChatMessage>? result, _, _) = await RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, "abc")]);
Assert.Equal("Hello from agent1", updateText);
Assert.NotNull(result);
@@ -296,7 +297,7 @@ public class AgentWorkflowBuilderTests
.WithHandoff(initialAgent, nextAgent)
.Build();
(string updateText, List<ChatMessage>? result, _) = await RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, "abc")]);
(string updateText, List<ChatMessage>? result, _, _) = await RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, "abc")]);
Assert.Equal("Hello from agent2", updateText);
Assert.NotNull(result);
@@ -406,7 +407,7 @@ public class AgentWorkflowBuilderTests
.WithHandoff(secondAgent, thirdAgent)
.Build();
(string updateText, _, _) = await RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, "abc")]);
(string updateText, _, _, _) = await RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, "abc")]);
Assert.Contains("Hello from agent3", updateText);
@@ -604,7 +605,7 @@ public class AgentWorkflowBuilderTests
.WithHandoff(secondAgent, thirdAgent)
.Build();
(string updateText, List<ChatMessage>? result, _) = await RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, "abc")]);
(string updateText, List<ChatMessage>? result, _, _) = await RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, "abc")]);
Assert.Equal("Hello from agent3", updateText);
Assert.NotNull(result);
@@ -634,6 +635,232 @@ public class AgentWorkflowBuilderTests
Assert.Contains("thirdAgent", result[5].AuthorName);
}
[Fact]
public async Task Handoffs_TwoTransfers_SecondAgentUserApproval_ResponseServedByThirdAgentAsync()
{
var initialAgent = new ChatClientAgent(new MockChatClient((messages, options) =>
{
ChatMessage message = Assert.Single(messages);
Assert.Equal("abc", Assert.IsType<TextContent>(Assert.Single(message.Contents)).Text);
string? transferFuncName = options?.Tools?.FirstOrDefault(t => t.Name.StartsWith("handoff_to_", StringComparison.Ordinal))?.Name;
Assert.NotNull(transferFuncName);
// Only a handoff function call.
return new(new ChatMessage(ChatRole.Assistant, [new FunctionCallContent("call1", transferFuncName)]));
}), name: "initialAgent");
bool secondAgentInvoked = false;
const string SomeOtherFunctionCallId = "call2first";
AIFunction someOtherFunction = new ApprovalRequiredAIFunction(AIFunctionFactory.Create(SomeOtherFunction));
var secondAgent = new ChatClientAgent(new MockChatClient((messages, options) =>
{
if (!secondAgentInvoked)
{
secondAgentInvoked = true;
return new(new ChatMessage(ChatRole.Assistant, [new FunctionCallContent(SomeOtherFunctionCallId, someOtherFunction.Name)]));
}
// Second agent should receive the conversation so far (including previous assistant + tool messages eventually).
string? transferFuncName = options?.Tools?.FirstOrDefault(t => t.Name.StartsWith("handoff_to_", StringComparison.Ordinal))?.Name;
Assert.NotNull(transferFuncName);
return new(new ChatMessage(ChatRole.Assistant, [new FunctionCallContent("call2", transferFuncName)]));
}), name: "secondAgent", description: "The second agent", tools: [someOtherFunction]);
var thirdAgent = new ChatClientAgent(new MockChatClient((messages, options) =>
new(new ChatMessage(ChatRole.Assistant, "Hello from agent3"))),
name: "thirdAgent",
description: "The third / final agent");
var workflow =
AgentWorkflowBuilder.CreateHandoffBuilderWith(initialAgent)
.WithHandoff(initialAgent, secondAgent)
.WithHandoff(secondAgent, thirdAgent)
.Build();
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
const ExecutionEnvironment Environment = ExecutionEnvironment.InProcess_Lockstep;
(string updateText, List<ChatMessage>? result, CheckpointInfo? lastCheckpoint, List<RequestInfoEvent> requests) =
await RunWorkflowCheckpointedAsync(workflow, [new ChatMessage(ChatRole.User, "abc")], Environment, checkpointManager);
Assert.Null(result);
Assert.NotNull(requests);
requests.Should().HaveCount(1);
ExternalRequest request = requests[0].Request;
ToolApprovalRequestContent approvalRequest =
request.Data.As<ToolApprovalRequestContent>().Should().NotBeNull()
.And.Subject.As<ToolApprovalRequestContent>();
approvalRequest.ToolCall.CallId.Should().Be(SomeOtherFunctionCallId);
ExternalResponse response = request.CreateResponse(approvalRequest.CreateResponse(false, "Denied"));
(updateText, result, _, requests) =
await RunWorkflowCheckpointedAsync(workflow, response, Environment, checkpointManager, lastCheckpoint);
Assert.Equal("Hello from agent3", updateText);
Assert.NotNull(result);
// User + (assistant empty + tool) for each of first two agents + final assistant with text.
Assert.Equal(10, result.Count);
Assert.Equal(ChatRole.User, result[0].Role);
Assert.Equal("abc", result[0].Text);
Assert.Equal(ChatRole.Assistant, result[1].Role);
Assert.Equal("", result[1].Text);
Assert.Contains("initialAgent", result[1].AuthorName);
Assert.Equal(ChatRole.Tool, result[2].Role);
Assert.Contains("initialAgent", result[2].AuthorName);
// Non-handoff tool invocation (and user denial)
Assert.Equal(ChatRole.Assistant, result[3].Role);
Assert.Equal("", result[3].Text);
Assert.Contains("secondAgent", result[3].AuthorName);
Assert.Equal(ChatRole.User, result[4].Role);
Assert.Equal("", result[4].Text);
// Rejected tool call
Assert.Equal(ChatRole.Assistant, result[5].Role);
Assert.Equal("", result[5].Text);
Assert.Contains("secondAgent", result[5].AuthorName);
Assert.Equal(ChatRole.Tool, result[6].Role);
Assert.Contains("secondAgent", result[6].AuthorName);
// Handoff invocation
Assert.Equal(ChatRole.Assistant, result[7].Role);
Assert.Equal("", result[7].Text);
Assert.Contains("secondAgent", result[7].AuthorName);
Assert.Equal(ChatRole.Tool, result[8].Role);
Assert.Contains("secondAgent", result[8].AuthorName);
Assert.Equal(ChatRole.Assistant, result[9].Role);
Assert.Equal("Hello from agent3", result[9].Text);
Assert.Contains("thirdAgent", result[9].AuthorName);
static bool SomeOtherFunction() => true;
}
[Fact]
public async Task Handoffs_TwoTransfers_SecondAgentToolCall_ResponseServedByThirdAgentAsync()
{
var initialAgent = new ChatClientAgent(new MockChatClient((messages, options) =>
{
ChatMessage message = Assert.Single(messages);
Assert.Equal("abc", Assert.IsType<TextContent>(Assert.Single(message.Contents)).Text);
string? transferFuncName = options?.Tools?.FirstOrDefault(t => t.Name.StartsWith("handoff_to_", StringComparison.Ordinal))?.Name;
Assert.NotNull(transferFuncName);
// Only a handoff function call.
return new(new ChatMessage(ChatRole.Assistant, [new FunctionCallContent("call1", transferFuncName)]));
}), name: "initialAgent");
bool secondAgentInvoked = false;
const string SomeOtherFunctionName = "SomeOtherFunction";
const string SomeOtherFunctionCallId = "call2first";
JsonElement otherFunctionSchema = AIFunctionFactory.Create(() => true).JsonSchema;
AIFunctionDeclaration someOtherFunction = AIFunctionFactory.CreateDeclaration(SomeOtherFunctionName, "Another function", otherFunctionSchema);
var secondAgent = new ChatClientAgent(new MockChatClient((messages, options) =>
{
if (!secondAgentInvoked)
{
secondAgentInvoked = true;
return new(new ChatMessage(ChatRole.Assistant, [new FunctionCallContent(SomeOtherFunctionCallId, SomeOtherFunctionName)]));
}
// Second agent should receive the conversation so far (including previous assistant + tool messages eventually).
string? transferFuncName = options?.Tools?.FirstOrDefault(t => t.Name.StartsWith("handoff_to_", StringComparison.Ordinal))?.Name;
Assert.NotNull(transferFuncName);
return new(new ChatMessage(ChatRole.Assistant, [new FunctionCallContent("call2", transferFuncName)]));
}), name: "secondAgent", description: "The second agent", tools: [someOtherFunction]);
var thirdAgent = new ChatClientAgent(new MockChatClient((messages, options) =>
new(new ChatMessage(ChatRole.Assistant, "Hello from agent3"))),
name: "thirdAgent",
description: "The third / final agent");
var workflow =
AgentWorkflowBuilder.CreateHandoffBuilderWith(initialAgent)
.WithHandoff(initialAgent, secondAgent)
.WithHandoff(secondAgent, thirdAgent)
.Build();
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
const ExecutionEnvironment Environment = ExecutionEnvironment.InProcess_Lockstep;
(string updateText, List<ChatMessage>? result, CheckpointInfo? lastCheckpoint, List<RequestInfoEvent> requests) =
await RunWorkflowCheckpointedAsync(workflow, [new ChatMessage(ChatRole.User, "abc")], Environment, checkpointManager);
Assert.Null(result);
Assert.NotNull(requests);
requests.Should().HaveCount(1);
ExternalRequest request = requests[0].Request;
FunctionCallContent functionCall = request.Data.As<FunctionCallContent>().Should().NotBeNull()
.And.Subject.As<FunctionCallContent>();
functionCall.CallId.Should().Be(SomeOtherFunctionCallId);
functionCall.Name.Should().Be(SomeOtherFunctionName);
ExternalResponse response = request.CreateResponse(new FunctionResultContent(functionCall.CallId, true));
(updateText, result, _, requests) =
await RunWorkflowCheckpointedAsync(workflow, response, Environment, checkpointManager, lastCheckpoint);
Assert.Equal("Hello from agent3", updateText);
Assert.NotNull(result);
// User + (assistant empty + tool) for each of first two agents + final assistant with text.
Assert.Equal(8, result.Count);
Assert.Equal(ChatRole.User, result[0].Role);
Assert.Equal("abc", result[0].Text);
Assert.Equal(ChatRole.Assistant, result[1].Role);
Assert.Equal("", result[1].Text);
Assert.Contains("initialAgent", result[1].AuthorName);
Assert.Equal(ChatRole.Tool, result[2].Role);
Assert.Contains("initialAgent", result[2].AuthorName);
// Non-handoff tool invocation
Assert.Equal(ChatRole.Assistant, result[3].Role);
Assert.Equal("", result[3].Text);
Assert.Contains("secondAgent", result[3].AuthorName);
Assert.Equal(ChatRole.Tool, result[4].Role);
Assert.Contains("secondAgent", result[4].AuthorName);
// Handoff invocation
Assert.Equal(ChatRole.Assistant, result[5].Role);
Assert.Equal("", result[5].Text);
Assert.Contains("secondAgent", result[5].AuthorName);
Assert.Equal(ChatRole.Tool, result[6].Role);
Assert.Contains("secondAgent", result[6].AuthorName);
Assert.Equal(ChatRole.Assistant, result[7].Role);
Assert.Equal("Hello from agent3", result[7].Text);
Assert.Contains("thirdAgent", result[7].AuthorName);
}
[Theory]
[InlineData(1)]
[InlineData(2)]
@@ -651,7 +878,7 @@ public class AgentWorkflowBuilderTests
for (int iter = 0; iter < 3; iter++)
{
const string UserInput = "abc";
(string updateText, List<ChatMessage>? result, _) = await RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, UserInput)]);
(string updateText, List<ChatMessage>? result, _, _) = await RunWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, UserInput)]);
Assert.NotNull(result);
Assert.Equal(maxIterations + 1, result.Count);
@@ -832,7 +1059,7 @@ public class AgentWorkflowBuilderTests
Assert.Equal(1, specialistCallCount); // specialist NOT called
}
private sealed record WorkflowRunResult(string UpdateText, List<ChatMessage>? Result, CheckpointInfo? LastCheckpoint);
private sealed record WorkflowRunResult(string UpdateText, List<ChatMessage>? Result, CheckpointInfo? LastCheckpoint, List<RequestInfoEvent> PendingRequests);
private static Task<WorkflowRunResult> RunWorkflowCheckpointedAsync(
Workflow workflow, List<ChatMessage> input, ExecutionEnvironment executionEnvironment, CheckpointManager checkpointManager, CheckpointInfo? fromCheckpoint = null)
@@ -843,6 +1070,15 @@ public class AgentWorkflowBuilderTests
return RunWorkflowCheckpointedAsync(workflow, input, environment, fromCheckpoint);
}
private static Task<WorkflowRunResult> RunWorkflowCheckpointedAsync(
Workflow workflow, ExternalResponse response, ExecutionEnvironment executionEnvironment, CheckpointManager checkpointManager, CheckpointInfo? fromCheckpoint = null)
{
InProcessExecutionEnvironment environment = executionEnvironment.ToWorkflowExecutionEnvironment()
.WithCheckpointing(checkpointManager);
return RunWorkflowCheckpointedAsync(workflow, response, environment, fromCheckpoint);
}
private static async Task<WorkflowRunResult> RunWorkflowCheckpointedAsync(
Workflow workflow, List<ChatMessage> input, InProcessExecutionEnvironment environment, CheckpointInfo? fromCheckpoint = null)
{
@@ -853,15 +1089,39 @@ public class AgentWorkflowBuilderTests
await run.TrySendMessageAsync(input);
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
return await ProcessWorkflowRunAsync(run);
}
private static async Task<WorkflowRunResult> RunWorkflowCheckpointedAsync(
Workflow workflow, ExternalResponse response, InProcessExecutionEnvironment environment, CheckpointInfo? fromCheckpoint = null)
{
await using StreamingRun run =
fromCheckpoint != null ? await environment.ResumeStreamingAsync(workflow, fromCheckpoint)
: await environment.OpenStreamingAsync(workflow);
await run.SendResponseAsync(response);
return await ProcessWorkflowRunAsync(run);
}
private static async Task<WorkflowRunResult> ProcessWorkflowRunAsync(StreamingRun run)
{
StringBuilder sb = new();
WorkflowOutputEvent? output = null;
CheckpointInfo? lastCheckpoint = null;
await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
List<RequestInfoEvent> pendingRequests = [];
await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false).ConfigureAwait(false))
{
switch (evt)
{
case AgentResponseUpdateEvent executorComplete:
sb.Append(executorComplete.Data);
case AgentResponseUpdateEvent responseUpdate:
sb.Append(responseUpdate.Data);
break;
case RequestInfoEvent requestInfo:
pendingRequests.Add(requestInfo);
break;
case WorkflowOutputEvent e:
@@ -878,7 +1138,7 @@ public class AgentWorkflowBuilderTests
}
}
return new(sb.ToString(), output?.As<List<ChatMessage>>(), lastCheckpoint);
return new(sb.ToString(), output?.As<List<ChatMessage>>(), lastCheckpoint, pendingRequests);
}
private static Task<WorkflowRunResult> RunWorkflowAsync(
dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointResumeTests.cs
+42
View File
@@ -279,6 +279,48 @@ public class CheckpointResumeTests
"the workflow should be able to continue after the runtime restore replay");
}
/// <summary>
/// Verifies that restoring a live run clears any queued external responses from the
/// superseded timeline before importing checkpoint state.
/// </summary>
[Fact]
internal async Task Checkpoint_Restore_ClearsQueuedExternalResponsesBeforeImportAsync()
{
Workflow workflow = CreateSimpleRequestWorkflow();
CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
InProcessExecutionEnvironment env = ExecutionEnvironment.InProcess_Lockstep.ToWorkflowExecutionEnvironment();
await using StreamingRun run = await env.WithCheckpointing(checkpointManager)
.RunStreamingAsync(workflow, "Hello");
(ExternalRequest pendingRequest, CheckpointInfo checkpoint) = await CapturePendingRequestAndCheckpointAsync(run);
await run.SendResponseAsync(pendingRequest.CreateResponse("World"));
await run.RestoreCheckpointAsync(checkpoint);
List<WorkflowEvent> restoredEvents = await ReadToHaltAsync(run);
ExternalRequest replayedRequest = restoredEvents.OfType<RequestInfoEvent>()
.Select(evt => evt.Request)
.Should()
.ContainSingle("the restored run should still be waiting for the checkpointed request")
.Subject;
restoredEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
"a queued response from the superseded timeline should not be processed after restore");
RunStatus statusAfterRestore = await run.GetStatusAsync();
statusAfterRestore.Should().Be(RunStatus.PendingRequests,
"the restored run should remain pending until a post-restore response is sent");
await run.SendResponseAsync(replayedRequest.CreateResponse("Again"));
List<WorkflowEvent> completionEvents = await ReadToHaltAsync(run);
completionEvents.OfType<WorkflowErrorEvent>().Should().BeEmpty(
"the restored request should complete cleanly once a new response is provided");
RunStatus finalStatus = await run.GetStatusAsync();
finalStatus.Should().Be(RunStatus.Idle,
"the workflow should finish once the replayed request receives a fresh response");
}
/// <summary>
/// Verifies that a resumed parent workflow re-emits pending requests that originated in a subworkflow.
/// </summary>
dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/HandoffAgentExecutorTests.cs
+2 -2
View File
@@ -29,7 +29,7 @@ public class HandoffAgentExecutorTests : AIAgentHostingExecutorTestsBase
emitAgentResponseUpdateEvents: executorSetting,
HandoffToolCallFilteringBehavior.None);
HandoffAgentExecutor executor = new(agent, options);
HandoffAgentExecutor executor = new(agent, [], options);
testContext.ConfigureExecutor(executor);
// Act
@@ -57,7 +57,7 @@ public class HandoffAgentExecutorTests : AIAgentHostingExecutorTestsBase
emitAgentResponseUpdateEvents: false,
HandoffToolCallFilteringBehavior.None);
HandoffAgentExecutor executor = new(agent, options);
HandoffAgentExecutor executor = new(agent, [], options);
testContext.ConfigureExecutor(executor);
// Act