Round 2 of cleanup for agent runtime and orchestrations (#164)

- Moved InProcessRuntime type into abstractions package and deleted InProcess package.
- Moved several members of IAgentRuntime to be extension methods instead, e.g. multiple GetActorAsync overloads.
- Added synchronous RegisterMessageHandler overloads and used them to avoid unnecessary async usage at call sites.
- Removed unnecessary surface area from InProcessRuntime, e.g. StopAsync, RunUntilIdleAsync, etc.
- Fixed spin loop in InProcessRuntime that would consume an entire core for the duration of the orchestration's operation.
- Removed a bunch of allocation from InProcessRuntime.
- Made a runtime optional for orchestrations, defaulting to using a temporary InProcessRuntime if none is provided.
- Removed custom delegate types from orchestrations.
- Consolidated namespaces.
- Used records to simplify message classes.
- Tweaked naming on AgentActor to make purpose of protected methods more clear.
- Removed invocation in AgentActor.InvokeAsync of empty update / isFinal parameter.
- Changed OrchestrationHandoffs to avoid needing to pass in agents duplicatively.
- Made various extension methods, such as those on OrchestrationHandoffsExtensions, into instance methods.
This commit is contained in:
Stephen Toub
2025-07-11 11:12:18 -04:00
committed by GitHub
Unverified
parent eca0eb9cd0
commit 456e7d1b65
83 changed files with 939 additions and 1664 deletions
+1
View File
@@ -13,6 +13,7 @@
<PackageVersion Include="System.Linq.Async" Version="6.0.1" />
<PackageVersion Include="System.Text.Json" Version="8.0.6" />
<PackageVersion Include="System.Diagnostics.DiagnosticSource" Version="9.0.7" />
<PackageVersion Include="System.Threading.Channels" Version="9.0.7" />
<PackageVersion Include="System.Threading.Tasks.Extensions" Version="4.6.3" />
<!-- Microsoft.Extensions.* -->
<PackageVersion Include="Microsoft.Bcl.HashCode" Version="6.0.0" />
-4
View File
@@ -115,7 +115,6 @@
<Project Path="src/Microsoft.Extensions.AI.Agents.Abstractions/Microsoft.Extensions.AI.Agents.Abstractions.csproj" />
<Project Path="src/Microsoft.Extensions.AI.Agents.AzureAI/Microsoft.Extensions.AI.Agents.AzureAI.csproj" />
<Project Path="src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/Microsoft.Extensions.AI.Agents.Runtime.Abstractions.csproj" />
<Project Path="src/Microsoft.Extensions.AI.Agents.Runtime.InProcess/Microsoft.Extensions.AI.Agents.Runtime.InProcess.csproj" />
<Project Path="src/Microsoft.Extensions.AI.Agents/Microsoft.Extensions.AI.Agents.csproj" />
<Project Path="src/Microsoft.Extensions.AI.Agents.CopilotStudio/Microsoft.Extensions.AI.Agents.CopilotStudio.csproj" />
</Folder>
@@ -129,9 +128,6 @@
<Project Path="tests/Microsoft.Extensions.AI.Agents.Runtime.Abstractions.UnitTests/Microsoft.Extensions.AI.Agents.Runtime.Abstractions.UnitTests.csproj">
<BuildType Solution="Publish|*" Project="Debug" />
</Project>
<Project Path="tests/Microsoft.Extensions.AI.Agents.Runtime.InProcess.UnitTests/Microsoft.Extensions.AI.Agents.Runtime.InProcess.UnitTests.csproj">
<BuildType Solution="Publish|*" Project="Debug" />
</Project>
<Project Path="tests/Microsoft.Extensions.AI.Agents.UnitTests/Microsoft.Extensions.AI.Agents.UnitTests.csproj">
<BuildType Solution="Publish|*" Project="Debug" />
</Project>
+1 -4
View File
@@ -159,11 +159,8 @@ public class AgentSample(ITestOutputHelper output) : BaseSample(output)
private async Task AzureAIAgentsPersistentAgentCleanUpAsync(ChatClientAgent agent, AgentThread? thread, CancellationToken cancellationToken)
{
var persistentAgentsClient = agent.ChatClient.GetService<PersistentAgentsClient>();
if (persistentAgentsClient is null)
{
var persistentAgentsClient = agent.ChatClient.GetService<PersistentAgentsClient>() ??
throw new InvalidOperationException("The provided chat client is not a Persistent Agents Chat Client");
}
await persistentAgentsClient.Administration.DeleteAgentAsync(agent.Id, cancellationToken);
@@ -1,9 +1,7 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.Orchestration;
using Microsoft.Agents.Orchestration.Concurrent;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime.InProcess;
namespace Orchestration;
@@ -42,20 +40,14 @@ public class ConcurrentOrchestration_Intro(ITestOutputHelper output) : Orchestra
StreamingResponseCallback = streamedResponse ? monitor.StreamingResultCallback : null,
};
// Start the runtime
await using InProcessRuntime runtime = new();
await runtime.StartAsync();
// Run the orchestration
string input = "What is temperature?";
Console.WriteLine($"\n# INPUT: {input}\n");
OrchestrationResult<string[]> result = await orchestration.InvokeAsync(input, runtime);
OrchestrationResult<string[]> result = await orchestration.InvokeAsync(input);
string[] output = await result.GetValueAsync(TimeSpan.FromSeconds(ResultTimeoutInSeconds));
string[] output = await result;
Console.WriteLine($"\n# RESULT:\n{string.Join("\n\n", output.Select(text => $"{text}"))}");
await runtime.RunUntilIdleAsync();
this.DisplayHistory(monitor.History);
}
}
@@ -2,10 +2,7 @@
using System.Text.Json;
using Microsoft.Agents.Orchestration;
using Microsoft.Agents.Orchestration.Concurrent;
using Microsoft.Agents.Orchestration.Transforms;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime.InProcess;
using Microsoft.Shared.Samples;
namespace Orchestration;
@@ -43,20 +40,14 @@ public class ConcurrentOrchestration_With_StructuredOutput(ITestOutputHelper out
ResultTransform = outputTransform.TransformAsync,
};
// Start the runtime
await using InProcessRuntime runtime = new();
await runtime.StartAsync();
// Run the orchestration
const string resourceId = "Hamlet_full_play_summary.txt";
string input = Resources.Read(resourceId);
Console.WriteLine($"\n# INPUT: @{resourceId}\n");
OrchestrationResult<Analysis> result = await orchestration.InvokeAsync(input, runtime);
OrchestrationResult<Analysis> result = await orchestration.InvokeAsync(input);
Analysis output = await result.GetValueAsync(TimeSpan.FromSeconds(ResultTimeoutInSeconds * 2));
Analysis output = await result;
Console.WriteLine($"\n# RESULT:\n{JsonSerializer.Serialize(output, s_options)}");
await runtime.RunUntilIdleAsync();
}
#pragma warning disable CA1812 // Avoid uninstantiated internal classes
@@ -1,9 +1,7 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.Orchestration;
using Microsoft.Agents.Orchestration.GroupChat;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime.InProcess;
namespace Orchestration;
@@ -68,17 +66,10 @@ public class GroupChatOrchestration_Intro(ITestOutputHelper output) : Orchestrat
StreamingResponseCallback = streamedResponse ? monitor.StreamingResultCallback : null,
};
// Start the runtime
await using InProcessRuntime runtime = new();
await runtime.StartAsync();
string input = "Create a slogon for a new eletric SUV that is affordable and fun to drive.";
Console.WriteLine($"\n# INPUT: {input}\n");
OrchestrationResult<string> result = await orchestration.InvokeAsync(input, runtime);
string text = await result.GetValueAsync(TimeSpan.FromSeconds(ResultTimeoutInSeconds * 3));
Console.WriteLine($"\n# RESULT: {text}");
await runtime.RunUntilIdleAsync();
OrchestrationResult<string> result = await orchestration.InvokeAsync(input);
Console.WriteLine($"\n# RESULT: {await result}");
this.DisplayHistory(monitor.History);
}
@@ -1,10 +1,8 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.Orchestration;
using Microsoft.Agents.Orchestration.GroupChat;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime.InProcess;
namespace Orchestration;
@@ -132,17 +130,10 @@ public class GroupChatOrchestration_With_AIManager(ITestOutputHelper output) : O
ResponseCallback = monitor.ResponseCallback,
};
// Start the runtime
await using InProcessRuntime runtime = new();
await runtime.StartAsync();
// Run the orchestration
Console.WriteLine($"\n# INPUT: {topic}\n");
OrchestrationResult<string> result = await orchestration.InvokeAsync(topic, runtime);
string text = await result.GetValueAsync(TimeSpan.FromSeconds(ResultTimeoutInSeconds * 3));
Console.WriteLine($"\n# RESULT: {text}");
await runtime.RunUntilIdleAsync();
OrchestrationResult<string> result = await orchestration.InvokeAsync(topic);
Console.WriteLine($"\n# RESULT: {await result}");
this.DisplayHistory(monitor.History);
}
@@ -1,10 +1,8 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.Orchestration;
using Microsoft.Agents.Orchestration.GroupChat;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime.InProcess;
namespace Orchestration;
@@ -68,18 +66,11 @@ public class GroupChatOrchestration_With_HumanInTheLoop(ITestOutputHelper output
ResponseCallback = monitor.ResponseCallback,
};
// Start the runtime
await using InProcessRuntime runtime = new();
await runtime.StartAsync();
// Run the orchestration
string input = "Create a slogon for a new eletric SUV that is affordable and fun to drive.";
Console.WriteLine($"\n# INPUT: {input}\n");
OrchestrationResult<string> result = await orchestration.InvokeAsync(input, runtime);
string text = await result.GetValueAsync(TimeSpan.FromSeconds(ResultTimeoutInSeconds * 3));
Console.WriteLine($"\n# RESULT: {text}");
await runtime.RunUntilIdleAsync();
OrchestrationResult<string> result = await orchestration.InvokeAsync(input);
Console.WriteLine($"\n# RESULT: {await result}");
this.DisplayHistory(monitor.History);
}
@@ -97,23 +88,12 @@ public class GroupChatOrchestration_With_HumanInTheLoop(ITestOutputHelper output
{
string? lastAgent = history.LastOrDefault()?.AuthorName;
GroupChatManagerResult<bool> result;
GroupChatManagerResult<bool> result =
lastAgent is null ? new(false) { Reason = "No agents have spoken yet." } :
lastAgent is "Reviewer" ? new(true) { Reason = "User input is needed after the reviewer's message." } :
new(false) { Reason = "User input is not needed until the reviewer's message." };
if (lastAgent is null)
{
result = new GroupChatManagerResult<bool>(false) { Reason = "No agents have spoken yet." };
}
if (lastAgent == "Reviewer")
{
result = new GroupChatManagerResult<bool>(true) { Reason = "User input is needed after the reviewer's message." };
}
else
{
result = new GroupChatManagerResult<bool>(false) { Reason = "User input is not needed until the reviewer's message." };
}
return new ValueTask<GroupChatManagerResult<bool>>(result);
return new(result);
}
}
}
@@ -1,10 +1,8 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.Orchestration;
using Microsoft.Agents.Orchestration.Handoff;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime.InProcess;
namespace Orchestration;
@@ -65,11 +63,7 @@ public class HandoffOrchestration_Intro(ITestOutputHelper output) : Orchestratio
.Add(triageAgent, statusAgent, returnAgent, refundAgent)
.Add(statusAgent, triageAgent, "Transfer to this agent if the issue is not status related")
.Add(returnAgent, triageAgent, "Transfer to this agent if the issue is not return related")
.Add(refundAgent, triageAgent, "Transfer to this agent if the issue is not refund related"),
triageAgent,
statusAgent,
returnAgent,
refundAgent)
.Add(refundAgent, triageAgent, "Transfer to this agent if the issue is not refund related"))
{
InteractiveCallback = () =>
{
@@ -84,18 +78,11 @@ public class HandoffOrchestration_Intro(ITestOutputHelper output) : Orchestratio
StreamingResponseCallback = streamedResponse ? monitor.StreamingResultCallback : null,
};
// Start the runtime
await using InProcessRuntime runtime = new();
await runtime.StartAsync();
// Run the orchestration
Console.WriteLine($"\n# INPUT:\n{task}\n");
OrchestrationResult<string> result = await orchestration.InvokeAsync(task, runtime);
OrchestrationResult<string> result = await orchestration.InvokeAsync(task);
string text = await result.GetValueAsync(TimeSpan.FromSeconds(300));
Console.WriteLine($"\n# RESULT: {text}");
await runtime.RunUntilIdleAsync();
Console.WriteLine($"\n# RESULT: {await result}");
this.DisplayHistory(monitor.History);
}
@@ -2,10 +2,8 @@
using System.Text.Json.Serialization;
using Microsoft.Agents.Orchestration;
using Microsoft.Agents.Orchestration.Handoff;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime.InProcess;
namespace Orchestration;
@@ -49,10 +47,7 @@ public class HandoffOrchestration_With_StructuredInput(ITestOutputHelper output)
HandoffOrchestration<GithubIssue, string> orchestration =
new(OrchestrationHandoffs
.StartWith(triageAgent)
.Add(triageAgent, dotnetAgent, pythonAgent),
triageAgent,
pythonAgent,
dotnetAgent)
.Add(triageAgent, dotnetAgent, pythonAgent))
{
LoggerFactory = this.LoggerFactory,
ResponseCallback = monitor.ResponseCallback,
@@ -83,18 +78,11 @@ public class HandoffOrchestration_With_StructuredInput(ITestOutputHelper output)
Labels = []
};
// Start the runtime
await using InProcessRuntime runtime = new();
await runtime.StartAsync();
// Run the orchestration
Console.WriteLine($"\n# INPUT:\n{input.Id}: {input.Title}\n");
OrchestrationResult<string> result = await orchestration.InvokeAsync(input, runtime);
string text = await result.GetValueAsync(TimeSpan.FromSeconds(ResultTimeoutInSeconds));
Console.WriteLine($"\n# RESULT: {text}");
OrchestrationResult<string> result = await orchestration.InvokeAsync(input);
Console.WriteLine($"\n# RESULT: {await result}");
Console.WriteLine($"\n# LABELS: {string.Join(",", githubPlugin.Labels["12345"])}\n");
await runtime.RunUntilIdleAsync();
}
private sealed class GithubIssue
@@ -1,9 +1,7 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.Orchestration;
using Microsoft.Agents.Orchestration.Sequential;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime.InProcess;
namespace Orchestration;
@@ -64,18 +62,11 @@ public class SequentialOrchestration_Intro(ITestOutputHelper output) : Orchestra
StreamingResponseCallback = streamedResponse ? monitor.StreamingResultCallback : null,
};
// Start the runtime
await using InProcessRuntime runtime = new();
await runtime.StartAsync();
// Run the orchestration
string input = "An eco-friendly stainless steel water bottle that keeps drinks cold for 24 hours";
Console.WriteLine($"\n# INPUT: {input}\n");
OrchestrationResult<string> result = await orchestration.InvokeAsync(input, runtime);
string text = await result.GetValueAsync(TimeSpan.FromSeconds(ResultTimeoutInSeconds));
Console.WriteLine($"\n# RESULT: {text}");
await runtime.RunUntilIdleAsync();
OrchestrationResult<string> result = await orchestration.InvokeAsync(input);
Console.WriteLine($"\n# RESULT: {await result}");
this.DisplayHistory(monitor.History);
}
@@ -1,9 +1,7 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.Orchestration;
using Microsoft.Agents.Orchestration.Sequential;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime.InProcess;
namespace Orchestration;
@@ -26,29 +24,22 @@ public class SequentialOrchestration_With_Cancellation(ITestOutputHelper output)
// Define the orchestration
SequentialOrchestration orchestration = new(agent) { LoggerFactory = this.LoggerFactory };
// Start the runtime
await using InProcessRuntime runtime = new();
await runtime.StartAsync();
// Run the orchestration
string input = "42";
Console.WriteLine($"\n# INPUT: {input}\n");
OrchestrationResult<string> result = await orchestration.InvokeAsync(input, runtime);
OrchestrationResult<string> result = await orchestration.InvokeAsync(input);
result.Cancel();
await Task.Delay(TimeSpan.FromSeconds(3));
try
{
string text = await result.GetValueAsync(TimeSpan.FromSeconds(ResultTimeoutInSeconds));
Console.WriteLine($"\n# RESULT: {text}");
Console.WriteLine($"\n# RESULT: {await result}");
}
catch (AggregateException exception)
catch (TimeoutException exception)
{
Console.WriteLine($"\n# CANCELLED: {exception.InnerException?.Message}");
Console.WriteLine($"\n# CANCELED: {exception.Message}");
}
await runtime.RunUntilIdleAsync();
}
}
@@ -1,4 +1,4 @@
// Copyright (c) Microsoft. All rights reserved.
// Copyright (c) Microsoft. All rights reserved.
/* This enables support for C# 9/10 records on older frameworks */
@@ -25,12 +25,7 @@ public abstract class AgentActor : OrchestrationActor
/// <param name="agent">An <see cref="Agent"/>.</param>
/// <param name="logger">The logger to use for the actor</param>
protected AgentActor(ActorId id, IAgentRuntime runtime, OrchestrationContext context, Agent agent, ILogger? logger = null)
: base(
id,
runtime,
context,
agent.Description,
logger)
: base(id, runtime, context, agent.Description, logger)
{
this.Agent = agent;
this.Thread = this.Agent.GetNewThread();
@@ -55,102 +50,75 @@ public abstract class AgentActor : OrchestrationActor
}
/// <summary>
/// Invokes the agent for a regular (not streamed) response.
/// Invokes the agent for a non-streamed responses.
/// </summary>
/// <param name="messages">The messages to send.</param>
/// <param name="options">The options for running the agent.</param>
/// <param name="cancellationToken">A cancellation token for the operation.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
/// <remarks>
/// Override this method to customize the invocation of the agent.
/// This method is not intended to be called directly; instead, use <see cref="RunAsync"/>.
/// This method exists to be overridden in derived classes in order to customize the invocation of the agent by <see cref="RunAsync"/>.
/// </remarks>
protected virtual Task<AgentRunResponse> InvokeAsync(
IReadOnlyCollection<ChatMessage> messages,
AgentRunOptions options,
CancellationToken cancellationToken = default) =>
this.Agent.RunAsync(
[.. messages],
this.Thread,
options,
cancellationToken);
protected virtual Task<AgentRunResponse> InvokeCoreAsync(
IReadOnlyCollection<ChatMessage> messages, AgentRunOptions? options, CancellationToken cancellationToken) =>
this.Agent.RunAsync([.. messages], this.Thread, options, cancellationToken);
/// <summary>
/// Invokes the agent for a streamed response.
/// Invokes the agent for a streamed responses.
/// </summary>
/// <param name="messages">The messages to send.</param>
/// <param name="options">The options for running the agent.</param>
/// <param name="cancellationToken">A cancellation token for the operation.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
/// <remarks>
/// Override this method to customize the invocation of the agent.
/// This method is not intended to be called directly; instead, use <see cref="RunAsync"/>.
/// This method exists to be overridden in derived classes in order to customize the invocation of the agent by <see cref="RunAsync"/>.
/// </remarks>
protected virtual IAsyncEnumerable<AgentRunResponseUpdate> InvokeStreamingAsync(IReadOnlyCollection<ChatMessage> messages, AgentRunOptions options, CancellationToken cancellationToken) =>
this.Agent.RunStreamingAsync(
messages,
this.Thread,
options,
cancellationToken);
protected virtual IAsyncEnumerable<AgentRunResponseUpdate> InvokeStreamingCoreAsync(
IReadOnlyCollection<ChatMessage> messages, AgentRunOptions? options, CancellationToken cancellationToken) =>
this.Agent.RunStreamingAsync(messages, this.Thread, options, cancellationToken);
/// <summary>
/// Invokes the agent with a single chat message.
/// This method sets the message role to <see cref="ChatRole.User"/> and delegates to the overload accepting multiple messages.
/// </summary>
/// <param name="input">The chat message content to send.</param>
/// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
/// <returns>A task that returns the response <see cref="ChatMessage"/>.</returns>
protected ValueTask<ChatMessage> InvokeAsync(ChatMessage input, CancellationToken cancellationToken) =>
this.InvokeAsync([input], cancellationToken);
/// <summary>
/// Invokes the agent with input messages and respond with both streamed and regular messages.
/// Runs the agent with input messages and respond with both streamed and regular messages.
/// </summary>
/// <param name="input">The list of chat messages to send.</param>
/// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
/// <returns>A task that returns the response <see cref="ChatMessage"/>.</returns>
protected async ValueTask<ChatMessage> InvokeAsync(IEnumerable<ChatMessage> input, CancellationToken cancellationToken)
protected async ValueTask<ChatMessage> RunAsync(IEnumerable<ChatMessage> input, CancellationToken cancellationToken)
{
this.Context.Cancellation.ThrowIfCancellationRequested();
using CancellationTokenSource combined = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, this.Context.CancellationToken);
cancellationToken = combined.Token;
AgentRunOptions options = new();
if (this.Context.StreamingResponseCallback == null)
// Utilize streaming iff a streaming callback is provided; otherwise, use the non-streaming API.
AgentRunResponse response;
if (this.Context.StreamingResponseCallback is { } streamingCallback)
{
// No need to utilize streaming if no callback is provided
AgentRunResponse response = await this.InvokeAsync([.. input], options, cancellationToken).ConfigureAwait(false);
// For streaming, enumerate all the updates, invoking the callback for each, and storing them all.
// Then convert them all into a single response instance.
List<AgentRunResponseUpdate> updates = [];
if (this.Context.ResponseCallback is not null)
await foreach (AgentRunResponseUpdate update in this.InvokeStreamingCoreAsync([.. input], options: null, cancellationToken).WithCancellation(this.Context.CancellationToken).ConfigureAwait(false))
{
await this.Context.ResponseCallback.Invoke(response.Messages).ConfigureAwait(false);
updates.Add(update);
await streamingCallback(update).ConfigureAwait(false);
}
return response.Messages.Last();
response = updates.ToAgentRunResponse();
}
IAsyncEnumerable<AgentRunResponseUpdate> streamedResponses = this.InvokeStreamingAsync([.. input], options, cancellationToken);
AgentRunResponseUpdate? lastStreamedResponse = null;
List<AgentRunResponseUpdate> updates = [];
await foreach (AgentRunResponseUpdate streamedResponse in streamedResponses.ConfigureAwait(false))
else
{
this.Context.Cancellation.ThrowIfCancellationRequested();
await HandleStreamedMessage(lastStreamedResponse, isFinal: false).ConfigureAwait(false);
lastStreamedResponse = streamedResponse;
// For non-streaming, just invoke the non-streaming method and get back the response.
response = await this.InvokeCoreAsync([.. input], options: null, cancellationToken).ConfigureAwait(false);
}
return updates.ToAgentRunResponse().Messages.Last();
async ValueTask HandleStreamedMessage(AgentRunResponseUpdate? streamedResponse, bool isFinal)
// Regardless of whether we invoked streaming callbacks for individual updates, invoke the non-streaming callback with the final response instance.
// This can be used as an indication of completeness if someone otherwise only cares about the streaming updates.
if (this.Context.ResponseCallback is { } responseCallback)
{
if (this.Context.StreamingResponseCallback != null && streamedResponse != null)
{
await this.Context.StreamingResponseCallback.Invoke(streamedResponse, isFinal).ConfigureAwait(false);
}
if (streamedResponse != null)
{
updates.Add(streamedResponse);
}
await responseCallback.Invoke(response.Messages).ConfigureAwait(false);
}
return response.Messages.LastOrDefault() ?? new();
}
}
@@ -4,7 +4,6 @@ using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Agents.Orchestration.Transforms;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
@@ -18,7 +17,7 @@ public abstract partial class AgentOrchestration<TInput, TOutput>
/// </summary>
private sealed class RequestActor : OrchestrationActor
{
private readonly OrchestrationInputTransform<TInput> _transform;
private readonly Func<TInput, CancellationToken, ValueTask<IEnumerable<ChatMessage>>> _transform;
private readonly Func<IEnumerable<ChatMessage>, ValueTask> _action;
private readonly TaskCompletionSource<TOutput> _completionSource;
@@ -36,7 +35,7 @@ public abstract partial class AgentOrchestration<TInput, TOutput>
ActorId id,
IAgentRuntime runtime,
OrchestrationContext context,
OrchestrationInputTransform<TInput> transform,
Func<TInput, CancellationToken, ValueTask<IEnumerable<ChatMessage>>> transform,
TaskCompletionSource<TOutput> completionSource,
Func<IEnumerable<ChatMessage>, ValueTask> action,
ILogger<RequestActor>? logger = null)
@@ -4,7 +4,6 @@ using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Agents.Orchestration.Transforms;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
@@ -19,8 +18,8 @@ public abstract partial class AgentOrchestration<TInput, TOutput>
private sealed class ResultActor<TResult> : OrchestrationActor
{
private readonly TaskCompletionSource<TOutput> _completionSource;
private readonly OrchestrationResultTransform<TResult> _transformResult;
private readonly OrchestrationOutputTransform<TOutput> _transform;
private readonly Func<TResult, IList<ChatMessage>> _transformResult;
private readonly Func<IList<ChatMessage>, CancellationToken, ValueTask<TOutput>> _transform;
/// <summary>
/// Initializes a new instance of the <see cref="AgentOrchestration{TInput, TOutput}.ResultActor{TResult}"/> class.
@@ -36,8 +35,8 @@ public abstract partial class AgentOrchestration<TInput, TOutput>
ActorId id,
IAgentRuntime runtime,
OrchestrationContext context,
OrchestrationResultTransform<TResult> transformResult,
OrchestrationOutputTransform<TOutput> transformOutput,
Func<TResult, IList<ChatMessage>> transformResult,
Func<IList<ChatMessage>, CancellationToken, ValueTask<TOutput>> transformOutput,
TaskCompletionSource<TOutput> completionSource,
ILogger<ResultActor<TResult>>? logger = null)
: base(id, runtime, context, $"{id.Type}_Actor", logger)
@@ -2,33 +2,20 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Agents.Orchestration.Extensions;
using Microsoft.Agents.Orchestration.Transforms;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.AI.Agents.Runtime.InProcess;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Shared.Diagnostics;
#pragma warning disable CA2000 // Dispose objects before losing scope
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// Called for every response is produced by any agent.
/// </summary>
/// <param name="response">The agent response</param>
public delegate ValueTask OrchestrationResponseCallback(IEnumerable<ChatMessage> response);
/// <summary>
/// Called to expose the streamed response produced by any agent.
/// </summary>
/// <param name="response">The agent response</param>
/// <param name="isFinal">Indicates if streamed content is final chunk of the message.</param>
public delegate ValueTask OrchestrationStreamingCallback(AgentRunResponseUpdate response, bool isFinal);
/// <summary>
/// Called when human interaction is requested.
/// </summary>
@@ -47,9 +34,17 @@ public abstract partial class AgentOrchestration<TInput, TOutput>
/// <param name="members">Specifies the member agents or orchestrations participating in this orchestration.</param>
protected AgentOrchestration(params Agent[] members)
{
_ = Throw.IfNull(members);
// Capture orchestration root name without generic parameters for use in
// agent type and topic formatting as well as logging.
this.OrchestrationLabel = this.GetType().Name.Split('`').First();
string name = this.GetType().Name;
int pos = name.IndexOf('`');
if (pos > 0)
{
name = name.Substring(0, pos);
}
this.OrchestrationLabel = name;
this.Members = members;
}
@@ -72,22 +67,22 @@ public abstract partial class AgentOrchestration<TInput, TOutput>
/// <summary>
/// Transforms the orchestration input into a source input suitable for processing.
/// </summary>
public OrchestrationInputTransform<TInput> InputTransform { get; init; } = DefaultTransforms.FromInput<TInput>;
public Func<TInput, CancellationToken, ValueTask<IEnumerable<ChatMessage>>> InputTransform { get; init; } = DefaultTransforms.FromInput<TInput>;
/// <summary>
/// Transforms the processed result into the final output form.
/// </summary>
public OrchestrationOutputTransform<TOutput> ResultTransform { get; init; } = DefaultTransforms.ToOutput<TOutput>;
public Func<IList<ChatMessage>, CancellationToken, ValueTask<TOutput>> ResultTransform { get; init; } = DefaultTransforms.ToOutput<TOutput>;
/// <summary>
/// Optional callback that is invoked for every agent response.
/// </summary>
public OrchestrationResponseCallback? ResponseCallback { get; init; }
public Func<IEnumerable<ChatMessage>, ValueTask>? ResponseCallback { get; init; }
/// <summary>
/// Optional callback that is invoked for every agent response.
/// Optional callback that is invoked for every agent update.
/// </summary>
public OrchestrationStreamingCallback? StreamingResponseCallback { get; init; }
public Func<AgentRunResponseUpdate, ValueTask>? StreamingResponseCallback { get; init; }
/// <summary>
/// Gets the list of member targets involved in the orchestration.
@@ -108,14 +103,17 @@ public abstract partial class AgentOrchestration<TInput, TOutput>
/// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
public async ValueTask<OrchestrationResult<TOutput>> InvokeAsync(
TInput input,
IAgentRuntime runtime,
IAgentRuntime? runtime = null,
CancellationToken cancellationToken = default)
{
Throw.IfNull(input, nameof(input));
TopicId topic = new($"{this.OrchestrationLabel}_{Guid.NewGuid().ToString().Replace("-", string.Empty)}");
cancellationToken.ThrowIfCancellationRequested();
TopicId topic = new($"{this.OrchestrationLabel}_{Guid.NewGuid():N}");
CancellationTokenSource orchestrationCancelSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
cancellationToken = orchestrationCancelSource.Token;
OrchestrationContext context =
new(this.OrchestrationLabel,
@@ -129,9 +127,10 @@ public abstract partial class AgentOrchestration<TInput, TOutput>
TaskCompletionSource<TOutput> completion = new();
ActorType orchestrationType = await this.RegisterAsync(runtime, context, completion, handoff: null).ConfigureAwait(false);
InProcessRuntime? temporaryRuntime = null;
runtime ??= temporaryRuntime = InProcessRuntime.StartNew();
cancellationToken.ThrowIfCancellationRequested();
ActorType orchestrationType = await this.RegisterAsync(runtime, context, completion, handoff: null).ConfigureAwait(false);
logger.LogOrchestrationInvoke(this.OrchestrationLabel, topic);
@@ -139,7 +138,7 @@ public abstract partial class AgentOrchestration<TInput, TOutput>
logger.LogOrchestrationYield(this.OrchestrationLabel, topic);
return new OrchestrationResult<TOutput>(context, completion, orchestrationCancelSource, logger);
return new OrchestrationResult<TOutput>(context, completion, orchestrationCancelSource, logger, temporaryRuntime);
}
/// <summary>
@@ -191,24 +190,22 @@ public abstract partial class AgentOrchestration<TInput, TOutput>
ActorType orchestrationEntry =
await runtime.RegisterOrchestrationAgentAsync(
this.FormatAgentType(context.Topic, "Boot"),
(agentId, runtime) =>
{
RequestActor actor =
new(agentId,
runtime,
context,
this.InputTransform,
completion,
StartAsync,
context.LoggerFactory.CreateLogger<RequestActor>());
return new ValueTask<IRuntimeActor>(actor);
}).ConfigureAwait(false);
(agentId, runtime) =>
{
RequestActor actor =
new(agentId,
runtime,
context,
this.InputTransform,
completion,
input => this.StartAsync(runtime, context.Topic, input, entryAgent),
context.LoggerFactory.CreateLogger<RequestActor>());
return new ValueTask<IRuntimeActor>(actor);
}).ConfigureAwait(false);
logger.LogOrchestrationRegistrationDone(context.Orchestration, context.Topic);
return orchestrationEntry;
ValueTask StartAsync(IEnumerable<ChatMessage> input) => this.StartAsync(runtime, context.Topic, input, entryAgent);
}
/// <summary>
@@ -219,12 +216,12 @@ public abstract partial class AgentOrchestration<TInput, TOutput>
IAgentRuntime runtime,
OrchestrationContext context,
TaskCompletionSource<TOutput> completion,
OrchestrationOutputTransform<TOutput> outputTransform)
Func<IList<ChatMessage>, CancellationToken, ValueTask<TOutput>> outputTransform)
{
/// <summary>
/// Register the final result type.
/// </summary>
public async ValueTask<ActorType> RegisterResultTypeAsync<TResult>(OrchestrationResultTransform<TResult> resultTransform)
public async ValueTask<ActorType> RegisterResultTypeAsync<TResult>(Func<TResult, IList<ChatMessage>> resultTransform)
{
// Register actor for final result
ActorType registeredType =
@@ -3,22 +3,19 @@
using System;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.AI.Agents.Runtime;
namespace Microsoft.Agents.Orchestration.Extensions;
namespace Microsoft.Extensions.AI.Agents.Runtime;
/// <summary>
/// Extension methods for <see cref="IAgentRuntime"/>.
/// </summary>
public static class RuntimeExtensions
internal static class AgentRuntimeExtensions
{
/// <summary>
/// Sends a message to the specified agent.
/// </summary>
public static async ValueTask PublishMessageAsync(this IAgentRuntime runtime, object message, ActorType agentType, CancellationToken cancellationToken = default)
{
await runtime.PublishMessageAsync(message, new TopicId(agentType.Name), sender: null, messageId: null, cancellationToken).ConfigureAwait(false);
}
public static ValueTask PublishMessageAsync(this IAgentRuntime runtime, object message, ActorType agentType, CancellationToken cancellationToken = default) =>
runtime.PublishMessageAsync(message, new TopicId(agentType.Name), sender: null, messageId: null, cancellationToken);
/// <summary>
/// Registers an agent factory for the specified agent type and associates it with the runtime.
@@ -42,10 +39,8 @@ public static class RuntimeExtensions
/// </summary>
/// <param name="runtime">The runtime for managing the subscription.</param>
/// <param name="agentType">The agent type to subscribe.</param>
public static async Task SubscribeAsync(this IAgentRuntime runtime, ActorType agentType)
{
await runtime.AddSubscriptionAsync(new TypeSubscription(agentType.Name, agentType)).ConfigureAwait(false);
}
public static Task SubscribeAsync(this IAgentRuntime runtime, ActorType agentType) =>
runtime.AddSubscriptionAsync(new TypeSubscription(agentType.Name, agentType)).AsTask();
/// <summary>
/// Subscribes the specified agent type to the provided topics.
@@ -7,7 +7,7 @@ using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
namespace Microsoft.Agents.Orchestration.Concurrent;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// An <see cref="AgentActor"/> used with the <see cref="ConcurrentOrchestration{TInput, TOutput}"/>.
@@ -37,10 +37,10 @@ internal sealed class ConcurrentActor : AgentActor
{
this.Logger.LogConcurrentAgentInvoke(this.Id);
ChatMessage response = await this.InvokeAsync(item.Messages, cancellationToken).ConfigureAwait(false);
ChatMessage response = await this.RunAsync(item.Messages, cancellationToken).ConfigureAwait(false);
this.Logger.LogConcurrentAgentResult(this.Id, response.Text);
await this.PublishMessageAsync(response.AsResultMessage(), this._handoffActor, cancellationToken).ConfigureAwait(false);
await this.PublishMessageAsync(new ConcurrentMessages.Result(response), this._handoffActor, cancellationToken).ConfigureAwait(false);
}
}
@@ -3,52 +3,20 @@
using System.Collections.Generic;
using Microsoft.Extensions.AI;
namespace Microsoft.Agents.Orchestration.Concurrent;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// Common messages used by the <see cref="ConcurrentOrchestration{TInput, TOutput}"/>.
/// </summary>
internal static class ConcurrentMessages
{
/// <summary>
/// An empty message instance as a default.
/// </summary>
public static readonly ChatMessage Empty = new();
/// <summary>
/// The input task for a <see cref="ConcurrentOrchestration{TInput, TOutput}"/>.
/// </summary>
public sealed class Request
{
/// <summary>
/// The request input.
/// </summary>
public IList<ChatMessage> Messages { get; init; } = [];
}
public sealed record Request(IList<ChatMessage> Messages);
/// <summary>
/// A result from a <see cref="ConcurrentOrchestration{TInput, TOutput}"/>.
/// </summary>
public sealed class Result
{
/// <summary>
/// The result message.
/// </summary>
public ChatMessage Message { get; init; } = Empty;
}
/// <summary>
/// Extension method to convert a <see cref="string"/> to a <see cref="Result"/>.
/// </summary>
public static Result AsResultMessage(this string text, ChatRole? role = null) => new() { Message = new ChatMessage(role ?? ChatRole.Assistant, text) };
/// <summary>
/// Extension method to convert a <see cref="ChatMessage"/> to a <see cref="Result"/>.
/// </summary>
public static Result AsResultMessage(this ChatMessage message) => new() { Message = message };
/// <summary>
/// Extension method to convert a collection of <see cref="ChatMessage"/> to a <see cref="ConcurrentMessages.Request"/>.
/// </summary>
public static Request AsInputMessage(this IEnumerable<ChatMessage> messages) => new() { Messages = [.. messages] };
public sealed record Result(ChatMessage Message);
}
@@ -5,7 +5,7 @@ using System.Linq;
using System.Threading.Tasks;
using Microsoft.Extensions.AI.Agents;
namespace Microsoft.Agents.Orchestration.Concurrent;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// An orchestration that broadcasts the input message to each agent.
@@ -3,13 +3,12 @@
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.Agents.Orchestration.Extensions;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
namespace Microsoft.Agents.Orchestration.Concurrent;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// An orchestration that broadcasts the input message to each agent.
@@ -32,7 +31,7 @@ public class ConcurrentOrchestration<TInput, TOutput>
/// <inheritdoc />
protected override ValueTask StartAsync(IAgentRuntime runtime, TopicId topic, IEnumerable<ChatMessage> input, ActorType? entryAgent)
{
return runtime.PublishMessageAsync(input.AsInputMessage(), topic);
return runtime.PublishMessageAsync(new ConcurrentMessages.Request([.. input]), topic);
}
/// <inheritdoc />
@@ -1,7 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics.CodeAnalysis;
using Microsoft.Agents.Orchestration.Concurrent;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
@@ -6,7 +6,7 @@ using System.Threading.Tasks;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
namespace Microsoft.Agents.Orchestration.Concurrent;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// Actor for capturing each <see cref="ConcurrentMessages.Result"/> message.
@@ -0,0 +1,73 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Generic;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.AI;
namespace Microsoft.Agents.Orchestration;
internal static class DefaultTransforms
{
public static ValueTask<IEnumerable<ChatMessage>> FromInput<TInput>(TInput input, CancellationToken cancellationToken = default) =>
new(input switch
{
IEnumerable<ChatMessage> messages => messages,
ChatMessage message => [message],
string text => [new ChatMessage(ChatRole.User, text)],
_ => [new ChatMessage(ChatRole.User, JsonSerializer.Serialize(input))]
});
public static ValueTask<TOutput> ToOutput<TOutput>(IList<ChatMessage> result, CancellationToken cancellationToken = default)
{
bool isSingleResult = result.Count == 1;
TOutput output =
GetDefaultOutput() ??
GetObjectOutput() ??
throw new InvalidOperationException($"Unable to transform output to {typeof(TOutput)}.");
return new ValueTask<TOutput>(output);
TOutput? GetObjectOutput()
{
if (isSingleResult)
{
try
{
return JsonSerializer.Deserialize<TOutput>(result[0].Text);
}
catch (JsonException)
{
}
}
return default;
}
TOutput? GetDefaultOutput()
{
if (typeof(TOutput).IsInstanceOfType(result))
{
return (TOutput)(object)result;
}
if (isSingleResult)
{
if (typeof(ChatMessage).IsAssignableFrom(typeof(TOutput)))
{
return (TOutput)(object)result[0];
}
if (typeof(string) == typeof(TOutput))
{
return (TOutput)(object)(result[0].Text ?? string.Empty);
}
}
return default;
}
}
}
@@ -8,7 +8,7 @@ using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
namespace Microsoft.Agents.Orchestration.GroupChat;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// An <see cref="AgentActor"/> used with the <see cref="GroupChatOrchestration{TInput, TOutput}"/>.
@@ -30,32 +30,20 @@ internal sealed class GroupChatAgentActor : AgentActor
{
this._cache = [];
this.RegisterMessageHandler<GroupChatMessages.Group>(this.HandleAsync);
this.RegisterMessageHandler<GroupChatMessages.Reset>(this.HandleAsync);
this.RegisterMessageHandler<GroupChatMessages.Group>((item, ctx) => this._cache.AddRange(item.Messages));
this.RegisterMessageHandler<GroupChatMessages.Reset>((item, ctx) => this.ResetThread());
this.RegisterMessageHandler<GroupChatMessages.Speak>(this.HandleAsync);
}
private ValueTask HandleAsync(GroupChatMessages.Group item, MessageContext messageContext, CancellationToken cancellationToken)
{
this._cache.AddRange(item.Messages);
return default;
}
private ValueTask HandleAsync(GroupChatMessages.Reset item, MessageContext messageContext, CancellationToken cancellationToken)
{
this.ResetThread();
return default;
}
private async ValueTask HandleAsync(GroupChatMessages.Speak item, MessageContext messageContext, CancellationToken cancellationToken)
{
this.Logger.LogChatAgentInvoke(this.Id);
ChatMessage response = await this.InvokeAsync(this._cache, cancellationToken).ConfigureAwait(false);
ChatMessage response = await this.RunAsync(this._cache, cancellationToken).ConfigureAwait(false);
this.Logger.LogChatAgentResult(this.Id, response.Text);
this._cache.Clear();
await this.PublishMessageAsync(response.AsGroupMessage(), this.Context.Topic, cancellationToken: cancellationToken).ConfigureAwait(false);
await this.PublishMessageAsync(new GroupChatMessages.Group([response]), this.Context.Topic, cancellationToken: cancellationToken).ConfigureAwait(false);
}
}
@@ -5,7 +5,7 @@ using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.AI;
namespace Microsoft.Agents.Orchestration.GroupChat;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// Represents the result of a group chat manager operation, including a value and a reason.
@@ -7,7 +7,7 @@ using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
namespace Microsoft.Agents.Orchestration.GroupChat;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// An <see cref="OrchestrationActor"/> used to manage a <see cref="GroupChatOrchestration{TInput, TOutput}"/>.
@@ -52,7 +52,7 @@ internal sealed class GroupChatManagerActor : OrchestrationActor
this._chat.AddRange(item.Messages);
await this.PublishMessageAsync(item.Messages.AsGroupMessage(), this.Context.Topic, cancellationToken: cancellationToken).ConfigureAwait(false);
await this.PublishMessageAsync(new GroupChatMessages.Group(item.Messages), this.Context.Topic, cancellationToken: cancellationToken).ConfigureAwait(false);
await this.ManageAsync(messageContext, cancellationToken).ConfigureAwait(false);
}
@@ -78,7 +78,7 @@ internal sealed class GroupChatManagerActor : OrchestrationActor
ChatMessage input = await this._manager.InteractiveCallback.Invoke().ConfigureAwait(false);
this.Logger.LogChatManagerUserInput(this.Id, input.Text);
this._chat.Add(input);
await this.PublishMessageAsync(input.AsGroupMessage(), this.Context.Topic, cancellationToken: cancellationToken).ConfigureAwait(false);
await this.PublishMessageAsync(new GroupChatMessages.Group([input]), this.Context.Topic, cancellationToken: cancellationToken).ConfigureAwait(false);
}
}
@@ -88,7 +88,7 @@ internal sealed class GroupChatManagerActor : OrchestrationActor
{
GroupChatManagerResult<string> filterResult = await this._manager.FilterResults(this._chat, cancellationToken).ConfigureAwait(false);
this.Logger.LogChatManagerResult(this.Id, filterResult.Value, filterResult.Reason);
await this.PublishMessageAsync(filterResult.Value.AsResultMessage(), this._orchestrationType, cancellationToken).ConfigureAwait(false);
await this.PublishMessageAsync(new GroupChatMessages.Result(new(ChatRole.Assistant, filterResult.Value)), this._orchestrationType, cancellationToken).ConfigureAwait(false);
return;
}
@@ -3,28 +3,17 @@
using System.Collections.Generic;
using Microsoft.Extensions.AI;
namespace Microsoft.Agents.Orchestration.GroupChat;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// Common messages used for agent chat patterns.
/// </summary>
public static class GroupChatMessages
internal static class GroupChatMessages
{
/// <summary>
/// An empty message instance as a default.
/// </summary>
internal static readonly ChatMessage Empty = new();
/// <summary>
/// Broadcast a message to all <see cref="GroupChatAgentActor"/>.
/// </summary>
public sealed class Group
{
/// <summary>
/// The chat message being broadcast.
/// </summary>
public IEnumerable<ChatMessage> Messages { get; init; } = [];
}
public sealed record Group(IEnumerable<ChatMessage> Messages);
/// <summary>
/// Reset/clear the conversation history for all <see cref="GroupChatAgentActor"/>.
@@ -34,13 +23,7 @@ public static class GroupChatMessages
/// <summary>
/// The final result.
/// </summary>
public sealed class Result
{
/// <summary>
/// The chat response message.
/// </summary>
public ChatMessage Message { get; init; } = Empty;
}
public sealed record Result(ChatMessage Message);
/// <summary>
/// Signal a <see cref="GroupChatAgentActor"/> to respond.
@@ -50,36 +33,11 @@ public static class GroupChatMessages
/// <summary>
/// The input task.
/// </summary>
public sealed class InputTask
public sealed record InputTask(IEnumerable<ChatMessage> Messages)
{
/// <summary>
/// A task that does not require any action.
/// Gets an input task that does not require any action.
/// </summary>
public static readonly InputTask None = new();
/// <summary>
/// The input that defines the task goal.
/// </summary>
public IEnumerable<ChatMessage> Messages { get; init; } = [];
public static InputTask None { get; } = new([]);
}
/// <summary>
/// Extension method to convert a <see cref="ChatMessage"/> to a <see cref="Group"/>.
/// </summary>
public static Group AsGroupMessage(this ChatMessage message) => new() { Messages = [message] };
/// <summary>
/// Extension method to convert a <see cref="ChatMessage"/> to a <see cref="Group"/>.
/// </summary>
public static Group AsGroupMessage(this IEnumerable<ChatMessage> messages) => new() { Messages = messages };
/// <summary>
/// Extension method to convert a <see cref="ChatMessage"/> to a <see cref="Result"/>.
/// </summary>
public static InputTask AsInputTaskMessage(this IEnumerable<ChatMessage> messages) => new() { Messages = messages };
/// <summary>
/// Extension method to convert a <see cref="ChatMessage"/> to a <see cref="Result"/>.
/// </summary>
public static Result AsResultMessage(this string text) => new() { Message = new(ChatRole.Assistant, text) };
}
@@ -2,7 +2,7 @@
using Microsoft.Extensions.AI.Agents;
namespace Microsoft.Agents.Orchestration.GroupChat;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// An orchestration that broadcasts the input message to each agent.
@@ -3,14 +3,13 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Microsoft.Agents.Orchestration.Extensions;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Orchestration.GroupChat;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// An orchestration that coordinates a group-chat.
@@ -42,7 +41,7 @@ public class GroupChatOrchestration<TInput, TOutput> :
{
throw new ArgumentException("Entry agent is not defined.", nameof(entryAgent));
}
return runtime.PublishMessageAsync(input.AsInputTaskMessage(), entryAgent.Value);
return runtime.PublishMessageAsync(new GroupChatMessages.InputTask(input), entryAgent.Value);
}
/// <inheritdoc />
@@ -1,7 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics.CodeAnalysis;
using Microsoft.Agents.Orchestration.GroupChat;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
@@ -4,29 +4,22 @@ using System;
using System.Collections.Generic;
using System.Linq;
namespace Microsoft.Agents.Orchestration.GroupChat;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// Describes a team of agents participating in a group chat.
/// </summary>
public class GroupChatTeam : Dictionary<string, (string Type, string Description)>;
/// <summary>
/// Extensions for <see cref="GroupChatTeam"/>.
/// </summary>
public static class ChatGroupExtensions
public sealed class GroupChatTeam : Dictionary<string, (string Type, string Description)>
{
/// <summary>
/// Format the names of the agents in the team as a comma delimimted list.
/// </summary>
/// <param name="team">The agent team</param>
/// <returns>A comma delimimted list of agent name.</returns>
public static string FormatNames(this GroupChatTeam team) => string.Join(",", team.Select(t => t.Key));
public string FormatNames() => string.Join(",", this.Select(t => t.Key));
/// <summary>
/// Format the names and descriptions of the agents in the team as a markdown list.
/// </summary>
/// <param name="team">The agent team</param>
/// <returns>A markdown list of agent names and descriptions.</returns>
public static string FormatList(this GroupChatTeam team) => string.Join(Environment.NewLine, team.Select(t => $"- {t.Key}: {t.Value.Description}"));
public string FormatList() => string.Join(Environment.NewLine, this.Select(t => $"- {t.Key}: {t.Value.Description}"));
}
@@ -6,7 +6,7 @@ using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.AI;
namespace Microsoft.Agents.Orchestration.GroupChat;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// A <see cref="GroupChatManager"/> that selects agents in a round-robin fashion.
@@ -9,7 +9,7 @@ using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
namespace Microsoft.Agents.Orchestration.Handoff;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// An actor used with the <see cref="HandoffOrchestration{TInput,TOutput}"/>.
@@ -54,48 +54,35 @@ internal sealed class HandoffActor : AgentActor
ToolMode = ChatToolMode.Auto
};
this.RegisterMessageHandler<HandoffMessages.InputTask>(this.HandleAsync);
this.RegisterMessageHandler<HandoffMessages.InputTask>(this.Handle);
this.RegisterMessageHandler<HandoffMessages.Request>(this.HandleAsync);
this.RegisterMessageHandler<HandoffMessages.Response>(this.HandleAsync);
this.RegisterMessageHandler<HandoffMessages.Response>(this.Handle);
}
/// <inheritdoc/>
protected override Task<AgentRunResponse> InvokeAsync(
IReadOnlyCollection<ChatMessage> messages,
AgentRunOptions options,
CancellationToken cancellationToken = default) =>
this._chatAgent.RunAsync(
[.. messages],
this.Thread,
options,
this._options,
cancellationToken);
protected override Task<AgentRunResponse> InvokeCoreAsync(
IReadOnlyCollection<ChatMessage> messages, AgentRunOptions? options, CancellationToken cancellationToken) =>
this._chatAgent.RunAsync([.. messages], this.Thread, options, this._options, cancellationToken);
/// <inheritdoc/>
protected override IAsyncEnumerable<AgentRunResponseUpdate> InvokeStreamingAsync(IReadOnlyCollection<ChatMessage> messages, AgentRunOptions options, CancellationToken cancellationToken) =>
this._chatAgent.RunStreamingAsync(
messages,
this.Thread,
options,
this._options,
cancellationToken);
protected override IAsyncEnumerable<AgentRunResponseUpdate> InvokeStreamingCoreAsync(
IReadOnlyCollection<ChatMessage> messages, AgentRunOptions? options, CancellationToken cancellationToken) =>
this._chatAgent.RunStreamingAsync(messages, this.Thread, options, this._options, cancellationToken);
/// <summary>
/// Gets or sets the callback to be invoked for interactive input.
/// </summary>
public OrchestrationInteractiveCallback? InteractiveCallback { get; init; }
private ValueTask HandleAsync(HandoffMessages.InputTask item, MessageContext messageContext, CancellationToken cancellationToken)
private void Handle(HandoffMessages.InputTask item, MessageContext messageContext)
{
this._taskSummary = null;
this._cache.AddRange(item.Messages);
return default;
}
private ValueTask HandleAsync(HandoffMessages.Response item, MessageContext messageContext, CancellationToken cancellationToken)
private void Handle(HandoffMessages.Response item, MessageContext messageContext)
{
this._cache.Add(item.Message);
return default;
}
private async ValueTask HandleAsync(HandoffMessages.Request item, MessageContext messageContext, CancellationToken cancellationToken)
@@ -109,7 +96,7 @@ internal sealed class HandoffActor : AgentActor
ChatMessage response;
try
{
response = await this.InvokeAsync(this._cache, cancellationToken).ConfigureAwait(false);
response = await this.RunAsync(this._cache, cancellationToken).ConfigureAwait(false);
}
catch (Exception exception)
{
@@ -126,7 +113,7 @@ internal sealed class HandoffActor : AgentActor
// Since we don't want to publish that message, so we only publish if the response is an ASSISTANT message.
if (response.Role == ChatRole.Assistant)
{
await this.PublishMessageAsync(new HandoffMessages.Response { Message = response }, this.Context.Topic, messageId: null, cancellationToken).ConfigureAwait(false);
await this.PublishMessageAsync(new HandoffMessages.Response(response), this.Context.Topic, messageId: null, cancellationToken).ConfigureAwait(false);
}
if (this._handoffAgent != null)
@@ -141,7 +128,7 @@ internal sealed class HandoffActor : AgentActor
if (this.InteractiveCallback != null && this._taskSummary == null)
{
ChatMessage input = await this.InteractiveCallback().ConfigureAwait(false);
await this.PublishMessageAsync(new HandoffMessages.Response { Message = input }, this.Context.Topic, messageId: null, cancellationToken).ConfigureAwait(false);
await this.PublishMessageAsync(new HandoffMessages.Response(input), this.Context.Topic, messageId: null, cancellationToken).ConfigureAwait(false);
this._cache.Add(input);
continue;
}
@@ -187,7 +174,7 @@ internal sealed class HandoffActor : AgentActor
{
this.Logger.LogHandoffSummary(this.Id, summary);
this._taskSummary = summary;
await this.PublishMessageAsync(new HandoffMessages.Result { Message = new ChatMessage(ChatRole.Assistant, summary) }, this._resultHandoff, cancellationToken).ConfigureAwait(false);
await this.PublishMessageAsync(new HandoffMessages.Result(new(ChatRole.Assistant, summary)), this._resultHandoff, cancellationToken).ConfigureAwait(false);
if (FunctionInvokingChatClient.CurrentContext is not null)
{
@@ -3,39 +3,22 @@
using System.Collections.Generic;
using Microsoft.Extensions.AI;
namespace Microsoft.Agents.Orchestration.Handoff;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// A message that describes the input task and captures results for a <see cref="HandoffOrchestration{TInput,TOutput}"/>.
/// </summary>
internal static class HandoffMessages
{
/// <summary>
/// An empty message instance as a default.
/// </summary>
internal static readonly ChatMessage Empty = new();
/// <summary>
/// The input message.
/// </summary>
public sealed class InputTask
{
/// <summary>
/// The orchestration input messages.
/// </summary>
public IList<ChatMessage> Messages { get; init; } = [];
}
public sealed record InputTask(IList<ChatMessage> Messages);
/// <summary>
/// The final result.
/// </summary>
public sealed class Result
{
/// <summary>
/// The orchestration result message.
/// </summary>
public ChatMessage Message { get; init; } = Empty;
}
public sealed record Result(ChatMessage Message);
/// <summary>
/// Signals the handoff to another agent.
@@ -45,21 +28,5 @@ internal static class HandoffMessages
/// <summary>
/// Broadcast an agent response to all actors in the orchestration.
/// </summary>
public sealed class Response
{
/// <summary>
/// The chat response message.
/// </summary>
public ChatMessage Message { get; init; } = Empty;
}
/// <summary>
/// Extension method to convert a <see cref="ChatMessage"/> to a <see cref="Result"/>.
/// </summary>
public static InputTask AsInputTaskMessage(this IEnumerable<ChatMessage> messages) => new() { Messages = [.. messages] };
/// <summary>
/// Extension method to convert a <see cref="ChatMessage"/> to a <see cref="Result"/>.
/// </summary>
public static Result AsResultMessage(this ChatMessage message) => new() { Message = message };
public sealed record Response(ChatMessage Message);
}
@@ -2,7 +2,7 @@
using Microsoft.Extensions.AI.Agents;
namespace Microsoft.Agents.Orchestration.Handoff;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// An orchestration that passes the input message to the first agent, and
@@ -4,13 +4,12 @@ using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.Agents.Orchestration.Extensions;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
namespace Microsoft.Agents.Orchestration.Handoff;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// An orchestration that provides the input message to the first agent
@@ -24,17 +23,21 @@ public class HandoffOrchestration<TInput, TOutput> : AgentOrchestration<TInput,
/// Initializes a new instance of the <see cref="HandoffOrchestration{TInput, TOutput}"/> class.
/// </summary>
/// <param name="handoffs">Defines the handoff connections for each agent.</param>
/// <param name="agents">The agents participating in the orchestration.</param>
public HandoffOrchestration(OrchestrationHandoffs handoffs, params Agent[] agents)
: base(agents)
/// <param name="agents">Additional agents participating in the orchestration that weren't passed to <paramref name="handoffs"/>.</param>
public HandoffOrchestration(OrchestrationHandoffs handoffs, params Agent[] agents) : base(
agents is { Length: 0 } ? handoffs.Agents.ToArray() :
handoffs.Agents is { Count: 0 } ? agents :
handoffs.Agents.Concat(agents).Distinct().ToArray())
{
// Create list of distinct agent names
HashSet<string> agentNames = new(agents.Select(a => a.Name ?? a.Id), StringComparer.Ordinal)
HashSet<string> agentNames = new(base.Members.Select(a => a.Name ?? a.Id), StringComparer.Ordinal)
{
handoffs.FirstAgentName
};
// Extract names from handoffs that don't align with a member agent.
string[] badNames = [.. handoffs.Keys.Concat(handoffs.Values.SelectMany(h => h.Keys)).Where(name => !agentNames.Contains(name))];
// Fail fast if invalid names are present.
if (badNames.Length > 0)
{
@@ -56,7 +59,7 @@ public class HandoffOrchestration<TInput, TOutput> : AgentOrchestration<TInput,
{
throw new ArgumentException("Entry agent is not defined.", nameof(entryAgent));
}
await runtime.PublishMessageAsync(input.AsInputTaskMessage(), topic).ConfigureAwait(false);
await runtime.PublishMessageAsync(new HandoffMessages.InputTask([.. input]), topic).ConfigureAwait(false);
await runtime.PublishMessageAsync(new HandoffMessages.Request(), entryAgent.Value).ConfigureAwait(false);
}
@@ -1,7 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics.CodeAnalysis;
using Microsoft.Agents.Orchestration.Handoff;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
@@ -6,7 +6,7 @@ using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Orchestration.Handoff;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// Defines the handoff relationships for a given agent.
@@ -38,7 +38,9 @@ public sealed class OrchestrationHandoffs : Dictionary<string, AgentHandoffs>
/// <param name="firstAgent">The first agent to be invoked (prior to any handoff).</param>
public OrchestrationHandoffs(Agent firstAgent)
: this(firstAgent.Name ?? firstAgent.Id)
{ }
{
this.Agents.Add(firstAgent);
}
/// <summary>
/// Initializes a new instance of the <see cref="OrchestrationHandoffs"/> class with no handoff relationships.
@@ -62,26 +64,19 @@ public sealed class OrchestrationHandoffs : Dictionary<string, AgentHandoffs>
/// <param name="source">The source agent.</param>
/// <returns>The updated <see cref="OrchestrationHandoffs"/> instance.</returns>
public static OrchestrationHandoffs StartWith(Agent source) => new(source);
}
/// <summary>
/// Extension methods for building and modifying <see cref="OrchestrationHandoffs"/> relationships.
/// </summary>
public static class OrchestrationHandoffsExtensions
{
/// <summary>
/// Adds handoff relationships from a source agent to one or more target agents.
/// Each target agent's name or ID is mapped to its description.
/// </summary>
/// <param name="handoffs">The orchestration handoffs collection to update.</param>
/// <param name="source">The source agent.</param>
/// <param name="targets">The target agents to add as handoff targets for the source agent.</param>
/// <returns>The updated <see cref="OrchestrationHandoffs"/> instance.</returns>
public static OrchestrationHandoffs Add(this OrchestrationHandoffs handoffs, Agent source, params Agent[] targets)
public OrchestrationHandoffs Add(Agent source, params Agent[] targets)
{
string key = source.Name ?? source.Id;
AgentHandoffs agentHandoffs = handoffs.GetAgentHandoffs(key);
AgentHandoffs agentHandoffs = this.GetAgentHandoffs(key);
foreach (Agent target in targets)
{
@@ -90,60 +85,61 @@ public static class OrchestrationHandoffsExtensions
throw new InvalidOperationException($"The provided target agent with Id '{target.Id}' has no description or name, and no handoff description has been provided. At least one of these are required to register a handoff so that the appropriate target agent can be chosen.");
}
this.Agents.Add(target);
agentHandoffs[target.Name ?? target.Id] = target.Description ?? target.Name!;
}
return handoffs;
this.Agents.Add(source);
return this;
}
/// <summary>
/// Adds a handoff relationship from a source agent to a target agent with a custom description.
/// </summary>
/// <param name="handoffs">The orchestration handoffs collection to update.</param>
/// <param name="source">The source agent.</param>
/// <param name="target">The target agent.</param>
/// <param name="description">The handoff description.</param>
/// <returns>The updated <see cref="OrchestrationHandoffs"/> instance.</returns>
public static OrchestrationHandoffs Add(this OrchestrationHandoffs handoffs, Agent source, Agent target, string description)
=> handoffs.Add(source.Name ?? source.Id, target.Name ?? target.Id, description);
public OrchestrationHandoffs Add(Agent source, Agent target, string description)
=> this.Add(source.Name ?? source.Id, target.Name ?? target.Id, description);
/// <summary>
/// Adds a handoff relationship from a source agent to a target agent name/ID with a custom description.
/// </summary>
/// <param name="handoffs">The orchestration handoffs collection to update.</param>
/// <param name="source">The source agent.</param>
/// <param name="targetName">The target agent's name or ID.</param>
/// <param name="description">The handoff description.</param>
/// <returns>The updated <see cref="OrchestrationHandoffs"/> instance.</returns>
public static OrchestrationHandoffs Add(this OrchestrationHandoffs handoffs, Agent source, string targetName, string description)
=> handoffs.Add(source.Name ?? source.Id, targetName, description);
public OrchestrationHandoffs Add(Agent source, string targetName, string description)
=> this.Add(source.Name ?? source.Id, targetName, description);
/// <summary>
/// Adds a handoff relationship from a source agent name/ID to a target agent name/ID with a custom description.
/// </summary>
/// <param name="handoffs">The orchestration handoffs collection to update.</param>
/// <param name="sourceName">The source agent's name or ID.</param>
/// <param name="targetName">The target agent's name or ID.</param>
/// <param name="description">The handoff description.</param>
/// <returns>The updated <see cref="OrchestrationHandoffs"/> instance.</returns>
public static OrchestrationHandoffs Add(this OrchestrationHandoffs handoffs, string sourceName, string targetName, string description)
public OrchestrationHandoffs Add(string sourceName, string targetName, string description)
{
AgentHandoffs agentHandoffs = handoffs.GetAgentHandoffs(sourceName);
AgentHandoffs agentHandoffs = this.GetAgentHandoffs(sourceName);
agentHandoffs[targetName] = description;
return handoffs;
return this;
}
private static AgentHandoffs GetAgentHandoffs(this OrchestrationHandoffs handoffs, string key)
private AgentHandoffs GetAgentHandoffs(string key)
{
if (!handoffs.TryGetValue(key, out AgentHandoffs? agentHandoffs))
if (!this.TryGetValue(key, out AgentHandoffs? agentHandoffs))
{
agentHandoffs = [];
handoffs[key] = agentHandoffs;
this[key] = agentHandoffs = [];
}
return agentHandoffs;
}
internal HashSet<Agent> Agents { get; } = [];
}
/// <summary>
@@ -1,62 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics.CodeAnalysis;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// Extensions for logging <see cref="OrchestrationResult{TValue}"/>.
/// </summary>
/// <remarks>
/// This extension uses the <see cref="LoggerMessageAttribute"/> to
/// generate logging code at compile time to achieve optimized code.
/// </remarks>
[ExcludeFromCodeCoverage]
internal static partial class OrchestrationResultLogMessages
{
/// <summary>
/// Logs <see cref="OrchestrationResult{TValue}"/> awaiting the orchestration.
/// </summary>
[LoggerMessage(
Level = LogLevel.Trace,
Message = "AWAIT {Orchestration}: {Topic}")]
public static partial void LogOrchestrationResultAwait(
this ILogger logger,
string orchestration,
TopicId topic);
/// <summary>
/// Logs <see cref="OrchestrationResult{TValue}"/> timeout while awaiting the orchestration.
/// </summary>
[LoggerMessage(
Level = LogLevel.Error,
Message = "TIMEOUT {Orchestration}: {Topic}")]
public static partial void LogOrchestrationResultTimeout(
this ILogger logger,
string orchestration,
TopicId topic);
/// <summary>
/// Logs <see cref="OrchestrationResult{TValue}"/> cancelled the orchestration.
/// </summary>
[LoggerMessage(
Level = LogLevel.Error,
Message = "CANCELLED {Orchestration}: {Topic}")]
public static partial void LogOrchestrationResultCancelled(
this ILogger logger,
string orchestration,
TopicId topic);
/// <summary>
/// Logs <see cref="OrchestrationResult{TValue}"/> the awaited the orchestration has completed.
/// </summary>
[LoggerMessage(
Level = LogLevel.Trace,
Message = "COMPLETE {Orchestration}: {Topic}")]
public static partial void LogOrchestrationResultComplete(
this ILogger logger,
string orchestration,
TopicId topic);
}
@@ -1,10 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
#if !NET5_0_OR_GREATER
using System.ComponentModel;
namespace System.Runtime.CompilerServices;
[EditorBrowsable(EditorBrowsableState.Never)]
internal static class IsExternalInit { }
#endif
@@ -10,6 +10,7 @@
<PropertyGroup>
<InjectSharedThrow>true</InjectSharedThrow>
<InjectIsExternalInitOnLegacy>true</InjectIsExternalInitOnLegacy>
</PropertyGroup>
<Import Project="$(RepoRoot)/dotnet/nuget/nuget-package.props" />
@@ -23,7 +24,6 @@
<ItemGroup>
<ProjectReference Include="..\Microsoft.Extensions.AI.Agents\Microsoft.Extensions.AI.Agents.csproj" />
<ProjectReference Include="..\Microsoft.Extensions.AI.Agents.Runtime.Abstractions\Microsoft.Extensions.AI.Agents.Runtime.Abstractions.csproj" />
<ProjectReference Include="..\Microsoft.Extensions.AI.Agents.Runtime.InProcess\Microsoft.Extensions.AI.Agents.Runtime.InProcess.csproj" />
</ItemGroup>
<ItemGroup>
@@ -33,11 +33,9 @@ public abstract class OrchestrationActor : RuntimeActor
/// <param name="agentType">The recipient agent's type.</param>
/// <param name="cancellationToken">A token used to cancel the operation if needed.</param>
/// <returns>The agent identifier, if it exists.</returns>
protected async ValueTask PublishMessageAsync(
protected ValueTask PublishMessageAsync(
object message,
ActorType agentType,
CancellationToken cancellationToken = default)
{
await base.PublishMessageAsync(message, new TopicId(agentType.Name), messageId: null, cancellationToken).ConfigureAwait(false);
}
CancellationToken cancellationToken = default) =>
base.PublishMessageAsync(message, new TopicId(agentType.Name), messageId: null, cancellationToken);
}
@@ -1,6 +1,11 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
@@ -14,17 +19,17 @@ public sealed class OrchestrationContext
internal OrchestrationContext(
string orchestration,
TopicId topic,
OrchestrationResponseCallback? responseCallback,
OrchestrationStreamingCallback? streamingCallback,
Func<IEnumerable<ChatMessage>, ValueTask>? responseCallback,
Func<AgentRunResponseUpdate, ValueTask>? streamingCallback,
ILoggerFactory loggerFactory,
CancellationToken cancellation)
CancellationToken cancellationToken)
{
this.Orchestration = orchestration;
this.Topic = topic;
this.ResponseCallback = responseCallback;
this.StreamingResponseCallback = streamingCallback;
this.LoggerFactory = loggerFactory;
this.Cancellation = cancellation;
this.CancellationToken = cancellationToken;
}
/// <summary>
@@ -43,7 +48,7 @@ public sealed class OrchestrationContext
/// <summary>
/// Gets the cancellation token that can be used to observe cancellation requests for the orchestration.
/// </summary>
public CancellationToken Cancellation { get; }
public CancellationToken CancellationToken { get; }
/// <summary>
/// Gets the associated logger factory for creating loggers within the orchestration context.
@@ -53,10 +58,10 @@ public sealed class OrchestrationContext
/// <summary>
/// Optional callback that is invoked for every agent response.
/// </summary>
public OrchestrationResponseCallback? ResponseCallback { get; }
public Func<IEnumerable<ChatMessage>, ValueTask>? ResponseCallback { get; }
/// <summary>
/// Optional callback that is invoked for every agent response.
/// </summary>
public OrchestrationStreamingCallback? StreamingResponseCallback { get; }
public Func<AgentRunResponseUpdate, ValueTask>? StreamingResponseCallback { get; }
}
@@ -1,6 +1,8 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.ComponentModel;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.AI.Agents.Runtime;
@@ -13,31 +15,39 @@ namespace Microsoft.Agents.Orchestration;
/// This class encapsulates the asynchronous completion of an orchestration process.
/// </summary>
/// <typeparam name="TValue">The type of the value produced by the orchestration.</typeparam>
public sealed class OrchestrationResult<TValue> : IDisposable
public sealed partial class OrchestrationResult<TValue> : IAsyncDisposable
{
private readonly OrchestrationContext _context;
private readonly CancellationTokenSource _cancelSource;
private readonly TaskCompletionSource<TValue> _completion;
private readonly ILogger _logger;
private readonly IAsyncDisposable? _additionalDisposable;
private bool _isDisposed;
internal OrchestrationResult(OrchestrationContext context, TaskCompletionSource<TValue> completion, CancellationTokenSource orchestrationCancelSource, ILogger logger)
internal OrchestrationResult(OrchestrationContext context, TaskCompletionSource<TValue> completion, CancellationTokenSource orchestrationCancelSource, ILogger logger, IAsyncDisposable? additionalDisposable = null)
{
this._cancelSource = orchestrationCancelSource;
this._context = context;
this._completion = completion;
this._logger = logger;
this._additionalDisposable = additionalDisposable;
}
/// <summary>
/// Releases all resources used by the <see cref="OrchestrationResult{TValue}"/> instance.
/// </summary>
public void Dispose()
public async ValueTask DisposeAsync()
{
if (!this._isDisposed)
{
this._cancelSource.Dispose();
this._isDisposed = true;
this._cancelSource.Dispose();
if (this._additionalDisposable is { } ad)
{
await ad.DisposeAsync().ConfigureAwait(false);
}
}
}
@@ -52,42 +62,9 @@ public sealed class OrchestrationResult<TValue> : IDisposable
public TopicId Topic => this._context.Topic;
/// <summary>
/// Asynchronously retrieves the orchestration result value.
/// If a timeout is specified, the method will throw a <see cref="TimeoutException"/>
/// if the orchestration does not complete within the allotted time.
/// Gets a task that represents the completion of the orchestration result.
/// </summary>
/// <param name="timeout">An optional <see cref="TimeSpan"/> representing the maximum wait duration.</param>
/// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
/// <returns>A <see cref="ValueTask{TValue}"/> representing the result of the orchestration.</returns>
/// <exception cref="ObjectDisposedException">Thrown if this instance has been disposed.</exception>
/// <exception cref="TimeoutException">Thrown if the orchestration does not complete within the specified timeout period.</exception>
public async ValueTask<TValue> GetValueAsync(TimeSpan? timeout = null, CancellationToken cancellationToken = default)
{
#if NET
ObjectDisposedException.ThrowIf(this._isDisposed, this);
#else
if (this._isDisposed)
{
throw new ObjectDisposedException(this.GetType().Name);
}
#endif
this._logger.LogOrchestrationResultAwait(this.Orchestration, this.Topic);
if (timeout.HasValue)
{
Task[] tasks = [this._completion.Task];
if (!Task.WaitAll(tasks, timeout.Value))
{
this._logger.LogOrchestrationResultTimeout(this.Orchestration, this.Topic);
throw new TimeoutException($"Orchestration did not complete within the allowed duration ({timeout}).");
}
}
this._logger.LogOrchestrationResultComplete(this.Orchestration, this.Topic);
return await this._completion.Task.ConfigureAwait(false);
}
public Task<TValue> Task => this._completion.Task;
/// <summary>
/// Cancel the orchestration associated with this result.
@@ -108,8 +85,21 @@ public sealed class OrchestrationResult<TValue> : IDisposable
}
#endif
this._logger.LogOrchestrationResultCancelled(this.Orchestration, this.Topic);
this.LogOrchestrationResultCanceled(this.Orchestration, this.Topic);
this._cancelSource.Cancel();
this._completion.SetCanceled();
}
/// <summary>Enable directly awaiting an <see cref="OrchestrationResult{TValue}"/> by using <see cref="Task"/>'s awaiter.</summary>
[EditorBrowsable(EditorBrowsableState.Never)]
public TaskAwaiter<TValue> GetAwaiter() => this.Task.GetAwaiter();
/// <summary>
/// Logs <see cref="OrchestrationResult{TValue}"/> canceled the orchestration.
/// </summary>
[LoggerMessage(
Level = LogLevel.Error,
Message = "CANCELED {Orchestration}: {Topic}")]
private partial void LogOrchestrationResultCanceled(
string orchestration,
TopicId topic);
}
@@ -8,7 +8,7 @@ using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
namespace Microsoft.Agents.Orchestration.Sequential;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// An actor used with the <see cref="SequentialOrchestration{TInput,TOutput}"/>.
@@ -48,10 +48,10 @@ internal sealed class SequentialActor : AgentActor
this.Logger.LogSequentialAgentInvoke(this.Id);
ChatMessage response = await this.InvokeAsync(input, cancellationToken).ConfigureAwait(false);
ChatMessage response = await this.RunAsync(input, cancellationToken).ConfigureAwait(false);
this.Logger.LogSequentialAgentResult(this.Id, response.Text);
await this.PublishMessageAsync(response.AsResponseMessage(), this._nextAgent, cancellationToken: cancellationToken).ConfigureAwait(false);
await this.PublishMessageAsync(new SequentialMessages.Response(response), this._nextAgent, cancellationToken: cancellationToken).ConfigureAwait(false);
}
}
@@ -3,58 +3,20 @@
using System.Collections.Generic;
using Microsoft.Extensions.AI;
namespace Microsoft.Agents.Orchestration.Sequential;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// A message that describes the input task and captures results for a <see cref="SequentialOrchestration{TInput,TOutput}"/>.
/// </summary>
internal static class SequentialMessages
{
/// <summary>
/// An empty message instance as a default.
/// </summary>
public static readonly ChatMessage Empty = new();
/// <summary>
/// Represents a request containing a sequence of chat messages to be processed by the sequential orchestration.
/// </summary>
public sealed class Request
{
/// <summary>
/// The request input.
/// </summary>
public IList<ChatMessage> Messages { get; init; } = [];
}
public sealed record Request(IList<ChatMessage> Messages);
/// <summary>
/// Represents a response containing the result message from the sequential orchestration.
/// </summary>
public sealed class Response
{
/// <summary>
/// The response message.
/// </summary>
public ChatMessage Message { get; init; } = Empty;
}
/// <summary>
/// Extension method to convert a <see cref="ChatMessage"/> to a <see cref="SequentialMessages.Request"/>.
/// </summary>
/// <param name="message">The chat message to include in the request.</param>
/// <returns>A <see cref="SequentialMessages.Request"/> containing the provided messages.</returns>
public static Request AsRequestMessage(this ChatMessage message) => new() { Messages = [message] };
/// <summary>
/// Extension method to convert a collection of <see cref="ChatMessage"/> to a <see cref="SequentialMessages.Request"/>.
/// </summary>
/// <param name="messages">The collection of chat messages to include in the request.</param>
/// <returns>A <see cref="SequentialMessages.Request"/> containing the provided messages.</returns>
public static Request AsRequestMessage(this IEnumerable<ChatMessage> messages) => new() { Messages = [.. messages] };
/// <summary>
/// Extension method to convert a <see cref="ChatMessage"/> to a <see cref="SequentialMessages.Response"/>.
/// </summary>
/// <param name="message">The chat message to include in the response.</param>
/// <returns>A <see cref="SequentialMessages.Response"/> containing the provided message.</returns>
public static Response AsResponseMessage(this ChatMessage message) => new() { Message = message };
public sealed record Response(ChatMessage Message);
}
@@ -2,7 +2,7 @@
using Microsoft.Extensions.AI.Agents;
namespace Microsoft.Agents.Orchestration.Sequential;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// An orchestration that passes the input message to the first agent, and
@@ -3,13 +3,12 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Microsoft.Agents.Orchestration.Extensions;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
namespace Microsoft.Agents.Orchestration.Sequential;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// An orchestration that provides the input message to the first agent
@@ -33,7 +32,7 @@ public class SequentialOrchestration<TInput, TOutput> : AgentOrchestration<TInpu
{
throw new ArgumentException("Entry agent is not defined.", nameof(entryAgent));
}
await runtime.PublishMessageAsync(input.AsRequestMessage(), entryAgent.Value).ConfigureAwait(false);
await runtime.PublishMessageAsync(new SequentialMessages.Request([.. input]), entryAgent.Value).ConfigureAwait(false);
}
/// <inheritdoc />
@@ -1,7 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics.CodeAnalysis;
using Microsoft.Agents.Orchestration.Sequential;
using Microsoft.Extensions.AI.Agents.Runtime;
using Microsoft.Extensions.Logging;
@@ -1,75 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Generic;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.AI;
namespace Microsoft.Agents.Orchestration.Transforms;
internal static class DefaultTransforms
{
public static ValueTask<IEnumerable<ChatMessage>> FromInput<TInput>(TInput input, CancellationToken cancellationToken = default)
{
return new ValueTask<IEnumerable<ChatMessage>>(TransformInput());
IEnumerable<ChatMessage> TransformInput() =>
input switch
{
IEnumerable<ChatMessage> messages => messages,
ChatMessage message => [message],
string text => [new ChatMessage(ChatRole.User, text)],
_ => [new ChatMessage(ChatRole.User, JsonSerializer.Serialize(input))]
};
}
public static ValueTask<TOutput> ToOutput<TOutput>(IList<ChatMessage> result, CancellationToken cancellationToken = default)
{
bool isSingleResult = result.Count == 1;
TOutput output =
GetDefaultOutput() ??
GetObjectOutput() ??
throw new InvalidOperationException($"Unable to transform output to {typeof(TOutput)}.");
return new ValueTask<TOutput>(output);
TOutput? GetObjectOutput()
{
if (!isSingleResult)
{
return default;
}
try
{
return JsonSerializer.Deserialize<TOutput>(result[0].Text);
}
catch (JsonException)
{
return default;
}
}
TOutput? GetDefaultOutput()
{
object? output = null;
if (typeof(TOutput).IsAssignableFrom(result.GetType()))
{
output = (object)result;
}
else if (isSingleResult && typeof(ChatMessage).IsAssignableFrom(typeof(TOutput)))
{
output = (object)result[0];
}
else if (isSingleResult && typeof(string) == typeof(TOutput))
{
output = result[0].Text ?? string.Empty;
}
return (TOutput?)output;
}
}
}
@@ -1,34 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.AI;
namespace Microsoft.Agents.Orchestration.Transforms;
/// <summary>
/// Delegate for transforming an input of type <typeparamref name="TInput"/> into a collection of <see cref="ChatMessage"/>.
/// This is typically used to convert user or system input into a format suitable for chat orchestration.
/// </summary>
/// <param name="input">The input object to transform.</param>
/// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
/// <returns>A <see cref="ValueTask{TResult}"/> containing an enumerable of <see cref="ChatMessage"/> representing the transformed input.</returns>
public delegate ValueTask<IEnumerable<ChatMessage>> OrchestrationInputTransform<TInput>(TInput input, CancellationToken cancellationToken = default);
/// <summary>
/// Delegate for transforming a <see cref="ChatMessage"/> into an output of type <typeparamref name="TOutput"/>.
/// This is typically used to convert a chat response into a desired output format.
/// </summary>
/// <param name="result">The result messages to transform.</param>
/// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
/// <returns>A <see cref="ValueTask{TResult}"/> containing the transformed output of type <typeparamref name="TOutput"/>.</returns>
public delegate ValueTask<TOutput> OrchestrationOutputTransform<TOutput>(IList<ChatMessage> result, CancellationToken cancellationToken = default);
/// <summary>
/// Delegate for transforming the internal result message for an orchestration into a <see cref="ChatMessage"/>.
/// </summary>
/// <typeparam name="TResult">The result message type</typeparam>
/// <param name="result">The result messages</param>
/// <returns>The orchestration result as a <see cref="ChatMessage"/>.</returns>
public delegate IList<ChatMessage> OrchestrationResultTransform<TResult>(TResult result);
@@ -7,7 +7,7 @@ using System.Threading.Tasks;
using Microsoft.Extensions.AI;
using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Orchestration.Transforms;
namespace Microsoft.Agents.Orchestration;
/// <summary>
/// Populates the target result type <see cref="ChatMessage"/> into a structured output.
@@ -22,11 +22,6 @@ public readonly struct ActorMetadata : IEquatable<ActorMetadata>
throw new ArgumentException("Invalid actor key.", nameof(key));
}
if (description is null)
{
throw new ArgumentNullException(nameof(description));
}
this.Type = type;
this.Key = key;
this.Description = description;
@@ -0,0 +1,71 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Shared.Diagnostics;
namespace Microsoft.Extensions.AI.Agents.Runtime;
/// <summary>
/// Provides extension methods for the agent runtime.
/// </summary>
public static class AgentRuntimeExtensions
{
/// <summary>
/// Retrieves an actor by its type.
/// </summary>
/// <param name="agentRuntime">The agent runtime.</param>
/// <param name="actorType">The type of the actor.</param>
/// <param name="key">An optional key to specify variations of the actor. Defaults to "default".</param>
/// <param name="lazy">If <c>true</c>, the actor is fetched lazily.</param>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>A task representing the asynchronous operation, returning the actor's ID.</returns>
public static ValueTask<ActorId> GetActorAsync(this IAgentRuntime agentRuntime, ActorType actorType, string? key = null, bool lazy = true, CancellationToken cancellationToken = default)
{
Throw.IfNull(agentRuntime);
return agentRuntime.GetActorAsync(actorType.Name, key, lazy, cancellationToken);
}
/// <summary>
/// Retrieves an actor by its string representation.
/// </summary>
/// <param name="agentRuntime">The agent runtime.</param>
/// <param name="actor">The string representation of the actor.</param>
/// <param name="key">An optional key to specify variations of the actor. Defaults to "default".</param>
/// <param name="lazy">If <c>true</c>, the actor is fetched lazily.</param>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>A task representing the asynchronous operation, returning the actor's ID.</returns>
public static ValueTask<ActorId> GetActorAsync(this IAgentRuntime agentRuntime, string actor, string? key = null, bool lazy = true, CancellationToken cancellationToken = default)
{
Throw.IfNull(agentRuntime);
return agentRuntime.GetActorAsync(new ActorId(actor, key ?? "default"), lazy, cancellationToken);
}
/// <summary>
/// Registers an actor factory with the runtime, associating it with a specific actor type.
/// </summary>
/// <typeparam name="TActor">The type of actor created by the factory.</typeparam>
/// <param name="agentRuntime">The agent runtime.</param>
/// <param name="type">The actor type to associate with the factory.</param>
/// <param name="factoryFunc">A function that asynchronously creates the actor instance.</param>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>A task representing the asynchronous operation, returning the registered actor type.</returns>
public static ValueTask<ActorType> RegisterActorFactoryAsync<TActor>(
this IAgentRuntime agentRuntime,
ActorType type,
Func<ActorId, IAgentRuntime, ValueTask<TActor>> factoryFunc,
CancellationToken cancellationToken = default)
where TActor : IRuntimeActor
{
Throw.IfNull(agentRuntime);
Throw.IfNull(factoryFunc);
return agentRuntime.RegisterActorFactoryAsync(
type,
async ValueTask<IRuntimeActor> (actorId, runtime) => await factoryFunc(actorId, runtime).ConfigureAwait(false),
cancellationToken);
}
}
@@ -47,26 +47,6 @@ public interface IAgentRuntime : ISaveState
/// <returns>A task representing the asynchronous operation, returning the actor's ID.</returns>
ValueTask<ActorId> GetActorAsync(ActorId actorId, bool lazy = true, CancellationToken cancellationToken = default);
/// <summary>
/// Retrieves an actor by its type.
/// </summary>
/// <param name="actorType">The type of the actor.</param>
/// <param name="key">An optional key to specify variations of the actor. Defaults to "default".</param>
/// <param name="lazy">If <c>true</c>, the actor is fetched lazily.</param>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>A task representing the asynchronous operation, returning the actor's ID.</returns>
ValueTask<ActorId> GetActorAsync(ActorType actorType, string key = "default", bool lazy = true, CancellationToken cancellationToken = default);
/// <summary>
/// Retrieves an actor by its string representation.
/// </summary>
/// <param name="actor">The string representation of the actor.</param>
/// <param name="key">An optional key to specify variations of the actor. Defaults to "default".</param>
/// <param name="lazy">If <c>true</c>, the actor is fetched lazily.</param>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>A task representing the asynchronous operation, returning the actor's ID.</returns>
ValueTask<ActorId> GetActorAsync(string actor, string key = "default", bool lazy = true, CancellationToken cancellationToken = default);
/// <summary>
/// Saves the state of an actor.
/// The result must be JSON serializable.
@@ -32,6 +32,6 @@ public interface IRuntimeActor : ISaveState
/// A task representing the asynchronous operation, returning a response to the message.
/// The response can be <c>null</c> if no reply is necessary.
/// </returns>
/// <exception cref="OperationCanceledException">Thrown if the message was cancelled.</exception>
/// <exception cref="OperationCanceledException">Thrown if the message was canceled.</exception>
ValueTask<object?> OnMessageAsync(object message, MessageContext messageContext, CancellationToken cancellationToken = default);
}
@@ -0,0 +1,425 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Text.Json;
using System.Text.Json.Serialization;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using Microsoft.Shared.Diagnostics;
#pragma warning disable CA1711 // Identifiers should not have incorrect suffix
namespace Microsoft.Extensions.AI.Agents.Runtime.InProcess;
/// <summary>Provides an in-process/in-memory implementation of the agent runtime.</summary>
public sealed partial class InProcessRuntime : IAgentRuntime, IAsyncDisposable
{
private static readonly UnboundedChannelOptions s_singleReaderOptions = new();
private readonly Dictionary<ActorType, Func<ActorId, IAgentRuntime, ValueTask<IRuntimeActor>>> _actorFactories = [];
private readonly Dictionary<string, ISubscriptionDefinition> _subscriptions = [];
private readonly Channel<MessageToProcess> _messages = Channel.CreateUnbounded<MessageToProcess>(s_singleReaderOptions);
private readonly CancellationTokenSource _shutdownTokenSource = new();
private Task? _messageDeliveryTask;
private int _remainingWork = 1; // initial count of 1 represents overall operation, decremented when shutting down.
private int _signaledCompletion = 0;
// Internal for testing purposes.
internal readonly Dictionary<ActorId, IRuntimeActor> _actorInstances = [];
/// <summary>Initializes a new instance of the in-memory runtime.</summary>
public InProcessRuntime() { }
/// <summary>Gets the number of pending work items.</summary>
/// <remarks>Internal for testing purposes.</remarks>
internal int MessageCountForTesting => this._remainingWork - (1 - this._signaledCompletion);
/// <summary>Creates and starts a new <see cref="InProcessRuntime"/> instance.</summary>
/// <returns>The started runtime.</returns>
public static InProcessRuntime StartNew()
{
InProcessRuntime runtime = new();
runtime.Start();
return runtime;
}
/// <summary>Starts the runtime.</summary>
/// <exception cref="InvalidOperationException">Thrown if the runtime is already started.</exception>
public void Start()
{
ThrowIfInvalid(this._signaledCompletion != 0 || this._messageDeliveryTask is not null, "Runtime was already started or shutdown.");
CancellationToken ct = this._shutdownTokenSource.Token;
this._messageDeliveryTask = Task.Run(() => this.RunAsync(ct));
}
/// <inheritdoc/>
public async ValueTask DisposeAsync()
{
if (Interlocked.Exchange(ref this._signaledCompletion, 1) == 0 && this._messageDeliveryTask is not null)
{
this.DecrementRemainingWork();
this._shutdownTokenSource.Cancel();
this._shutdownTokenSource.Dispose();
await this._messageDeliveryTask.ConfigureAwait(false);
}
}
/// <inheritdoc/>
public ValueTask PublishMessageAsync(object message, TopicId topic, ActorId? sender = null, string? messageId = null, CancellationToken cancellationToken = default)
{
MessageToProcess m = new(this, message, messageId, sender, topic, cancellationToken);
this.IncrementRemainingWork();
this._messages.Writer.TryWrite(m);
return new(m.ResultTcs.Task);
}
/// <inheritdoc/>
public ValueTask<object?> SendMessageAsync(object message, ActorId recipient, ActorId? sender = null, string? messageId = null, CancellationToken cancellationToken = default)
{
MessageToProcess m = new(this, message, messageId, sender, recipient, cancellationToken);
this.IncrementRemainingWork();
this._messages.Writer.TryWrite(m);
return new(m.ResultTcs.Task);
}
/// <inheritdoc/>
public async ValueTask<ActorId> GetActorAsync(ActorId actorId, bool lazy = true, CancellationToken cancellationToken = default)
{
if (!lazy)
{
await this.EnsureActorAsync(actorId, cancellationToken).ConfigureAwait(false);
}
return actorId;
}
/// <inheritdoc/>
public async ValueTask<ActorMetadata> GetActorMetadataAsync(ActorId actorId, CancellationToken cancellationToken = default)
{
IRuntimeActor actor = await this.EnsureActorAsync(actorId, cancellationToken).ConfigureAwait(false);
return actor.Metadata;
}
/// <inheritdoc/>
public async ValueTask<TActor> TryGetUnderlyingActorInstanceAsync<TActor>(ActorId actorId, CancellationToken cancellationToken = default) where TActor : IRuntimeActor
{
IRuntimeActor actor = await this.EnsureActorAsync(actorId, cancellationToken).ConfigureAwait(false);
if (actor is not TActor concreteActor)
{
throw new InvalidOperationException($"Actor with name {actorId.Type} is not of type {typeof(TActor).Name}.");
}
return concreteActor;
}
/// <inheritdoc/>
public async ValueTask LoadActorStateAsync(ActorId actorId, JsonElement state, CancellationToken cancellationToken = default)
{
IRuntimeActor actor = await this.EnsureActorAsync(actorId, cancellationToken).ConfigureAwait(false);
await actor.LoadStateAsync(state, cancellationToken).ConfigureAwait(false);
}
/// <inheritdoc/>
public async ValueTask<JsonElement> SaveActorStateAsync(ActorId actorId, CancellationToken cancellationToken = default)
{
IRuntimeActor actor = await this.EnsureActorAsync(actorId, cancellationToken).ConfigureAwait(false);
return await actor.SaveStateAsync(cancellationToken).ConfigureAwait(false);
}
/// <inheritdoc/>
public ValueTask AddSubscriptionAsync(ISubscriptionDefinition subscription, CancellationToken cancellationToken = default)
{
ThrowIfInvalid(this._subscriptions.ContainsKey(subscription.Id), "Subscription with the specified ID already exists.");
this._subscriptions.Add(subscription.Id, subscription);
return default;
}
/// <inheritdoc/>
public ValueTask RemoveSubscriptionAsync(string subscriptionId, CancellationToken cancellationToken = default)
{
ThrowIfInvalid(!this._subscriptions.ContainsKey(subscriptionId), "Subscription with the specified ID does not exist.");
this._subscriptions.Remove(subscriptionId);
return default;
}
/// <inheritdoc/>
public async ValueTask LoadStateAsync(JsonElement state, CancellationToken cancellationToken = default)
{
foreach (JsonProperty actorIdStr in state.EnumerateObject())
{
ActorId actorId = ActorId.Parse(actorIdStr.Name);
if (this._actorFactories.ContainsKey(actorId.Type))
{
IRuntimeActor actor = await this.EnsureActorAsync(actorId, cancellationToken).ConfigureAwait(false);
await actor.LoadStateAsync(actorIdStr.Value, cancellationToken).ConfigureAwait(false);
}
}
}
/// <inheritdoc/>
public async ValueTask<JsonElement> SaveStateAsync(CancellationToken cancellationToken = default)
{
Dictionary<string, JsonElement> state = [];
foreach (KeyValuePair<ActorId, IRuntimeActor> actor in this._actorInstances)
{
state[actor.Key.ToString()] = await actor.Value.SaveStateAsync(cancellationToken).ConfigureAwait(false);
}
return JsonSerializer.SerializeToElement(state, InProcessRuntimeContext.Default.DictionaryStringJsonElement);
}
/// <inheritdoc/>
public async ValueTask<ActorType> RegisterActorFactoryAsync(ActorType type, Func<ActorId, IAgentRuntime, ValueTask<IRuntimeActor>> factoryFunc, CancellationToken cancellationToken = default)
{
ThrowIfInvalid(this._actorFactories.ContainsKey(type), "Actor type already registered.");
this._actorFactories.Add(type, factoryFunc);
return type;
}
/// <inheritdoc/>
public async ValueTask<IdProxyActor?> TryGetActorProxyAsync(ActorId actorId, CancellationToken cancellationToken = default) =>
new(this, actorId);
private async Task RunAsync(CancellationToken cancellationToken)
{
try
{
Dictionary<long, Task> pendingTasks = [];
long currentId = 0;
await foreach (MessageToProcess message in this._messages.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
{
this.DecrementRemainingWork();
ValueTask processTask = message.InvokeAsync(cancellationToken);
if (!processTask.IsCompleted)
{
currentId++;
Task t = WaitAndRemoveAsync(currentId, processTask);
lock (pendingTasks)
{
if (!t.IsCompleted)
{
pendingTasks.Add(currentId, t);
}
}
async Task WaitAndRemoveAsync(long taskId, ValueTask processTask)
{
try
{
await processTask.ConfigureAwait(false);
}
finally
{
lock (pendingTasks)
{
pendingTasks.Remove(taskId);
}
}
}
}
}
await Task.WhenAll(pendingTasks.Values).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
// Ignore cancellation exceptions, as they are expected when the runtime is shutting down.
}
finally
{
foreach (var actor in this._actorInstances)
{
if (actor.Value is IAsyncDisposable closeableActor)
{
await closeableActor.DisposeAsync().ConfigureAwait(false);
}
}
}
}
private static readonly Func<MessageToProcess, CancellationToken, ValueTask<object?>> s_publishServicer =
async (MessageToProcess message, CancellationToken cancellationToken) =>
{
Debug.Assert(message.Topic.HasValue);
List<Exception>? exceptions = null;
TopicId topic = message.Topic!.Value;
foreach (KeyValuePair<string, ISubscriptionDefinition> subscription in message.Runtime._subscriptions)
{
if (!subscription.Value.Matches(topic))
{
continue;
}
try
{
using CancellationTokenSource combinedSource = CancellationTokenSource.CreateLinkedTokenSource(message.Cancellation, cancellationToken);
combinedSource.Token.ThrowIfCancellationRequested();
ActorId actorId = subscription.Value.MapToActor(topic);
ActorId? sender = message.Sender;
if (sender is null || sender != actorId)
{
IRuntimeActor actor = await message.Runtime.EnsureActorAsync(actorId, combinedSource.Token).ConfigureAwait(false);
await actor.OnMessageAsync(message.Message, new()
{
MessageId = message.MessageId,
Sender = sender,
Topic = topic,
}, combinedSource.Token).ConfigureAwait(false);
}
}
catch (Exception ex)
{
(exceptions ??= []).Add(ex);
}
}
if (exceptions is not null)
{
throw new AggregateException("One or more exceptions occurred while processing the message.", exceptions);
}
// This method is effectively void, with the result never being used. But it's typed the same as SendMessageServicerAsync
// in order to be able to share the same consuming code.
return null;
};
private static readonly Func<MessageToProcess, CancellationToken, ValueTask<object?>> s_sendServicer =
async (MessageToProcess message, CancellationToken cancellationToken) =>
{
Debug.Assert(message.Receiver.HasValue);
using CancellationTokenSource combinedSource = CancellationTokenSource.CreateLinkedTokenSource(message.Cancellation, cancellationToken);
IRuntimeActor actor = await message.Runtime.EnsureActorAsync(message.Receiver!.Value, combinedSource.Token).ConfigureAwait(false);
return await actor.OnMessageAsync(message.Message, new()
{
MessageId = message.MessageId,
Sender = message.Sender,
}, combinedSource.Token).ConfigureAwait(false);
};
private async ValueTask<IRuntimeActor> EnsureActorAsync(ActorId actorId, CancellationToken cancellationToken)
{
if (!this._actorInstances.TryGetValue(actorId, out IRuntimeActor? actor))
{
this._actorFactories.TryGetValue(actorId.Type, out Func<ActorId, IAgentRuntime, ValueTask<IRuntimeActor>>? factoryFunc);
ThrowIfInvalid(factoryFunc is null, "Actor with the specified name not found.");
actor = await factoryFunc(actorId, this).ConfigureAwait(false);
this._actorInstances.Add(actorId, actor);
}
return actor;
}
private void IncrementRemainingWork()
{
int current;
do
{
current = this._remainingWork;
ThrowIfInvalid(current <= 0, "Runtime has already shut down.");
}
while (Interlocked.CompareExchange(ref this._remainingWork, current + 1, current) != current);
}
private void DecrementRemainingWork()
{
int current;
do
{
current = this._remainingWork;
ThrowIfInvalid(current <= 0, "Runtime has already shut down.");
}
while (Interlocked.CompareExchange(ref this._remainingWork, current - 1, current) != current);
if (current == 1)
{
this._messages.Writer.TryComplete();
}
}
private static void ThrowIfInvalid([DoesNotReturnIf(true)] bool isInvalid, string message)
{
if (isInvalid)
{
throw new InvalidOperationException(message);
}
}
[JsonSerializable(typeof(Dictionary<string, JsonElement>))]
private sealed partial class InProcessRuntimeContext : JsonSerializerContext;
private sealed class MessageToProcess
{
public MessageToProcess(InProcessRuntime runtime, object message, string? messageId, ActorId? sender, ActorId receiver, CancellationToken cancellationToken) :
this(runtime, message, messageId, sender, s_sendServicer, cancellationToken)
{
this.Receiver = Throw.IfNull(receiver);
}
public MessageToProcess(InProcessRuntime runtime, object message, string? messageId, ActorId? sender, TopicId topic, CancellationToken cancellationToken) :
this(runtime, message, messageId, sender, s_publishServicer, cancellationToken)
{
this.Topic = Throw.IfNull(topic);
}
private MessageToProcess(InProcessRuntime runtime, object message, string? messageId, ActorId? sender, Func<MessageToProcess, CancellationToken, ValueTask<object?>> servicer, CancellationToken cancellationToken)
{
this.Runtime = runtime;
this.Message = message;
this.MessageId = messageId ?? Guid.NewGuid().ToString();
this.Sender = sender;
this.Servicer = servicer;
this.Cancellation = cancellationToken;
}
public InProcessRuntime Runtime { get; }
public object Message { get; }
public string MessageId { get; }
public ActorId? Sender { get; }
public TopicId? Topic { get; }
public ActorId? Receiver { get; }
public CancellationToken Cancellation { get; }
public TaskCompletionSource<object?> ResultTcs { get; } = new();
private Func<MessageToProcess, CancellationToken, ValueTask<object?>> Servicer { get; }
public async ValueTask InvokeAsync(CancellationToken cancellationToken)
{
try
{
this.ResultTcs.SetResult(await this.Servicer(this, cancellationToken).ConfigureAwait(false));
}
catch (OperationCanceledException exception)
{
this.ResultTcs.TrySetCanceled(exception.CancellationToken);
}
catch (Exception exception)
{
this.ResultTcs.SetException(exception);
}
}
}
}
@@ -8,6 +8,7 @@
</PropertyGroup>
<PropertyGroup>
<InjectSharedThrow>true</InjectSharedThrow>
<InjectDiagnosticAttributesOnLegacy>true</InjectDiagnosticAttributesOnLegacy>
<InjectTrimAttributesOnLegacy>true</InjectTrimAttributesOnLegacy>
</PropertyGroup>
@@ -15,7 +16,7 @@
<Import Project="$(RepoRoot)/dotnet/nuget/nuget-package.props" />
<ItemGroup>
<None Include="IRuntimeActor.cs" />
<InternalsVisibleTo Include="Microsoft.Extensions.AI.Agents.Runtime.Abstractions.UnitTests" />
</ItemGroup>
<ItemGroup>
@@ -25,6 +26,7 @@
<ItemGroup Condition="'$(TargetFrameworkIdentifier)' != '.NETCoreApp'">
<PackageReference Include="Microsoft.Bcl.HashCode" />
<PackageReference Include="System.Text.Json" />
<PackageReference Include="System.Threading.Channels" />
<PackageReference Include="System.Threading.Tasks.Extensions" />
<PackageReference Include="System.Diagnostics.DiagnosticSource" />
</ItemGroup>
@@ -8,6 +8,7 @@ using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Shared.Diagnostics;
namespace Microsoft.Extensions.AI.Agents.Runtime;
@@ -63,6 +64,24 @@ public abstract class RuntimeActor : IRuntimeActor
this.Metadata = new ActorMetadata(this.Id.Type, this.Id.Key, description);
}
/// <summary>Registers a handler for <typeparamref name="TInput"/>.</summary>
/// <typeparam name="TInput">The type of the input message for the handler.</typeparam>
/// <param name="messageHandler">The handler function that processes the message.</param>
/// <exception cref="InvalidOperationException">Thrown when a handler for the specified type is already registered.</exception>
/// <remarks>
/// The base implementation of <see cref="OnMessageAsync"/> will use these registered handlers to process incoming messages.
/// </remarks>
protected void RegisterMessageHandler<TInput>(Action<TInput, MessageContext> messageHandler)
{
_ = Throw.IfNull(messageHandler);
this.RegisterMessageHandler<TInput, object?>(async (input, ctx, cancellationToken) =>
{
messageHandler(input, ctx);
return null;
});
}
/// <summary>Registers a handler for <typeparamref name="TInput"/>.</summary>
/// <typeparam name="TInput">The type of the input message for the handler.</typeparam>
/// <param name="messageHandler">The handler function that processes the message.</param>
@@ -72,23 +91,28 @@ public abstract class RuntimeActor : IRuntimeActor
/// </remarks>
protected void RegisterMessageHandler<TInput>(Func<TInput, MessageContext, CancellationToken, ValueTask> messageHandler)
{
if (messageHandler is null)
{
throw new ArgumentNullException(nameof(messageHandler));
}
_ = Throw.IfNull(messageHandler);
if (this._handlerInvokers.ContainsKey(typeof(TInput)))
this.RegisterMessageHandler<TInput, object?>(async (input, ctx, cancellationToken) =>
{
throw new InvalidOperationException($"A handler for type {typeof(TInput)} is already registered.");
}
await messageHandler(input, ctx, cancellationToken).ConfigureAwait(false);
return null;
});
}
this._handlerInvokers.Add(
typeof(TInput),
async (message, messageContext, cancellationToken) =>
{
await messageHandler((TInput)message!, messageContext, cancellationToken).ConfigureAwait(false);
return null; // No return value for void handlers
});
/// <summary>Registers a handler for <typeparamref name="TInput"/>.</summary>
/// <typeparam name="TInput">The type of the input message for the handler.</typeparam>
/// <typeparam name="TOutput">The type of the output message for the handler.</typeparam>
/// <param name="messageHandler">The handler function that processes the message.</param>
/// <exception cref="InvalidOperationException">Thrown when a handler for the specified type is already registered.</exception>
/// <remarks>
/// The base implementation of <see cref="OnMessageAsync"/> will use these registered handlers to process incoming messages.
/// </remarks>
protected void RegisterMessageHandler<TInput, TOutput>(Func<TInput, MessageContext, TOutput> messageHandler)
{
_ = Throw.IfNull(messageHandler);
this.RegisterMessageHandler<TInput, object?>(async (input, ctx, cancellationToken) => messageHandler(input, ctx));
}
/// <summary>Registers a handler for <typeparamref name="TInput"/> that produces a <typeparamref name="TOutput"/>.</summary>
@@ -101,10 +125,7 @@ public abstract class RuntimeActor : IRuntimeActor
/// </remarks>
protected void RegisterMessageHandler<TInput, TOutput>(Func<TInput, MessageContext, CancellationToken, ValueTask<TOutput>> messageHandler)
{
if (messageHandler is null)
{
throw new ArgumentNullException(nameof(messageHandler));
}
_ = Throw.IfNull(messageHandler);
if (this._handlerInvokers.ContainsKey(typeof(TInput)))
{
@@ -113,11 +134,7 @@ public abstract class RuntimeActor : IRuntimeActor
this._handlerInvokers.Add(
typeof(TInput),
async (message, messageContext, cancellationToken) =>
{
TOutput? result = await messageHandler((TInput)message!, messageContext, cancellationToken).ConfigureAwait(false);
return (object?)result;
});
async (message, messageContext, cancellationToken) => await messageHandler((TInput)message!, messageContext, cancellationToken).ConfigureAwait(false));
}
/// <summary>
@@ -1,437 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text.Json;
using System.Text.Json.Serialization;
using System.Threading;
using System.Threading.Tasks;
namespace Microsoft.Extensions.AI.Agents.Runtime.InProcess;
/// <summary>
/// Provides an in-process/in-memory implementation of the agent runtime.
/// </summary>
public sealed partial class InProcessRuntime : IAgentRuntime, IAsyncDisposable
{
private readonly Dictionary<ActorType, Func<ActorId, IAgentRuntime, ValueTask<IRuntimeActor>>> _actorFactories = [];
private readonly Dictionary<string, ISubscriptionDefinition> _subscriptions = [];
private readonly ConcurrentQueue<MessageDelivery> _messageDeliveryQueue = new();
private CancellationTokenSource? _shutdownSource;
private CancellationTokenSource? _finishSource;
private Task _messageDeliveryTask = Task.CompletedTask;
private Func<bool> _shouldContinue = () => true;
// Exposed for testing purposes.
internal int _messageQueueCount;
internal readonly Dictionary<ActorId, IRuntimeActor> _actorInstances = [];
/// <summary>
/// Gets or sets a value indicating whether actors should receive messages they send themselves.
/// </summary>
public bool DeliverToSelf { get; set; }
/// <inheritdoc/>
public async ValueTask DisposeAsync()
{
await this.RunUntilIdleAsync().ConfigureAwait(false);
this._shutdownSource?.Dispose();
this._finishSource?.Dispose();
}
/// <summary>
/// Starts the runtime service.
/// </summary>
/// <param name="cancellationToken">Token to monitor for shutdown requests.</param>
/// <returns>A task representing the asynchronous operation.</returns>
/// <exception cref="InvalidOperationException">Thrown if the runtime is already started.</exception>
public Task StartAsync(CancellationToken cancellationToken = default)
{
if (this._shutdownSource != null)
{
throw new InvalidOperationException("Runtime is already running.");
}
this._shutdownSource = new CancellationTokenSource();
this._messageDeliveryTask = Task.Run(() => this.RunAsync(this._shutdownSource.Token), cancellationToken);
return Task.CompletedTask;
}
/// <summary>
/// Stops the runtime service.
/// </summary>
/// <param name="cancellationToken">Token to propagate when stopping the runtime.</param>
/// <returns>A task representing the asynchronous operation.</returns>
/// <exception cref="InvalidOperationException">Thrown if the runtime is in the process of stopping.</exception>
public Task StopAsync(CancellationToken cancellationToken = default)
{
if (this._shutdownSource != null)
{
if (this._finishSource != null)
{
throw new InvalidOperationException("Runtime is already stopping.");
}
this._finishSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
this._shutdownSource.Cancel();
}
return Task.CompletedTask;
}
/// <summary>
/// This will run until the message queue is empty and then stop the runtime.
/// </summary>
public async Task RunUntilIdleAsync(CancellationToken cancellationToken = default)
{
Func<bool> oldShouldContinue = this._shouldContinue;
this._shouldContinue = () => !this._messageDeliveryQueue.IsEmpty;
// TODO: Do we want detach semantics?
await this._messageDeliveryTask.ConfigureAwait(false);
this._shouldContinue = oldShouldContinue;
}
/// <inheritdoc/>
public ValueTask PublishMessageAsync(object message, TopicId topic, ActorId? sender = null, string? messageId = null, CancellationToken cancellationToken = default)
{
return this.ExecuteTracedAsync(async () =>
{
MessageDelivery delivery =
new MessageEnvelope(message, messageId, cancellationToken)
.WithSender(sender)
.ForPublish(topic, this.PublishMessageServicerAsync);
this._messageDeliveryQueue.Enqueue(delivery);
Interlocked.Increment(ref this._messageQueueCount);
await delivery.ResultTask.ConfigureAwait(false);
});
}
/// <inheritdoc/>
public async ValueTask<object?> SendMessageAsync(object message, ActorId recipient, ActorId? sender = null, string? messageId = null, CancellationToken cancellationToken = default)
{
return await this.ExecuteTracedAsync(async () =>
{
MessageDelivery delivery =
new MessageEnvelope(message, messageId, cancellationToken)
.WithSender(sender)
.ForSend(recipient, this.SendMessageServicerAsync);
this._messageDeliveryQueue.Enqueue(delivery);
Interlocked.Increment(ref this._messageQueueCount);
return await delivery.ResultTask.ConfigureAwait(false);
}).ConfigureAwait(false);
}
/// <inheritdoc/>
public async ValueTask<ActorId> GetActorAsync(ActorId actorId, bool lazy = true, CancellationToken cancellationToken = default)
{
if (!lazy)
{
await this.EnsureActorAsync(actorId, cancellationToken).ConfigureAwait(false);
}
return actorId;
}
/// <inheritdoc/>
public ValueTask<ActorId> GetActorAsync(ActorType actorType, string? key = null, bool lazy = true, CancellationToken cancellationToken = default)
=> this.GetActorAsync(actorType.Name, key, lazy, cancellationToken);
/// <inheritdoc/>
public ValueTask<ActorId> GetActorAsync(string actor, string? key = null, bool lazy = true, CancellationToken cancellationToken = default)
=> this.GetActorAsync(new ActorId(actor, key ?? "default"), lazy, cancellationToken);
/// <inheritdoc/>
public async ValueTask<ActorMetadata> GetActorMetadataAsync(ActorId actorId, CancellationToken cancellationToken = default)
{
IRuntimeActor actor = await this.EnsureActorAsync(actorId, cancellationToken).ConfigureAwait(false);
return actor.Metadata;
}
/// <inheritdoc/>
public async ValueTask<TActor> TryGetUnderlyingActorInstanceAsync<TActor>(ActorId actorId, CancellationToken cancellationToken = default) where TActor : IRuntimeActor
{
IRuntimeActor actor = await this.EnsureActorAsync(actorId, cancellationToken).ConfigureAwait(false);
if (actor is not TActor concreteActor)
{
throw new InvalidOperationException($"Actor with name {actorId.Type} is not of type {typeof(TActor).Name}.");
}
return concreteActor;
}
/// <inheritdoc/>
public async ValueTask LoadActorStateAsync(ActorId actorId, JsonElement state, CancellationToken cancellationToken = default)
{
IRuntimeActor actor = await this.EnsureActorAsync(actorId, cancellationToken).ConfigureAwait(false);
await actor.LoadStateAsync(state, cancellationToken).ConfigureAwait(false);
}
/// <inheritdoc/>
public async ValueTask<JsonElement> SaveActorStateAsync(ActorId actorId, CancellationToken cancellationToken = default)
{
IRuntimeActor actor = await this.EnsureActorAsync(actorId, cancellationToken).ConfigureAwait(false);
return await actor.SaveStateAsync(cancellationToken).ConfigureAwait(false);
}
/// <inheritdoc/>
public ValueTask AddSubscriptionAsync(ISubscriptionDefinition subscription, CancellationToken cancellationToken = default)
{
if (this._subscriptions.ContainsKey(subscription.Id))
{
throw new InvalidOperationException($"Subscription with id {subscription.Id} already exists.");
}
this._subscriptions.Add(subscription.Id, subscription);
return default;
}
/// <inheritdoc/>
public ValueTask RemoveSubscriptionAsync(string subscriptionId, CancellationToken cancellationToken = default)
{
if (!this._subscriptions.ContainsKey(subscriptionId))
{
throw new InvalidOperationException($"Subscription with id {subscriptionId} does not exist.");
}
this._subscriptions.Remove(subscriptionId);
return default;
}
/// <inheritdoc/>
public async ValueTask LoadStateAsync(JsonElement state, CancellationToken cancellationToken = default)
{
foreach (JsonProperty actorIdStr in state.EnumerateObject())
{
ActorId actorId = ActorId.Parse(actorIdStr.Name);
if (this._actorFactories.ContainsKey(actorId.Type))
{
IRuntimeActor actor = await this.EnsureActorAsync(actorId, cancellationToken).ConfigureAwait(false);
await actor.LoadStateAsync(actorIdStr.Value, cancellationToken).ConfigureAwait(false);
}
}
}
/// <inheritdoc/>
public async ValueTask<JsonElement> SaveStateAsync(CancellationToken cancellationToken = default)
{
Dictionary<string, JsonElement> state = [];
foreach (ActorId actorId in this._actorInstances.Keys)
{
JsonElement actorState = await this._actorInstances[actorId].SaveStateAsync(cancellationToken).ConfigureAwait(false);
state[actorId.ToString()] = actorState;
}
return JsonSerializer.SerializeToElement(state, InProcessRuntimeContext.Default.DictionaryStringJsonElement);
}
/// <summary>
/// Registers an actor factory with the runtime, associating it with a specific actor type.
/// </summary>
/// <typeparam name="TActor">The type of actor created by the factory.</typeparam>
/// <param name="type">The actor type to associate with the factory.</param>
/// <param name="factoryFunc">A function that asynchronously creates the actor instance.</param>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>A task representing the asynchronous operation, returning the registered actor type.</returns>
public ValueTask<ActorType> RegisterActorFactoryAsync<TActor>(ActorType type, Func<ActorId, IAgentRuntime, ValueTask<TActor>> factoryFunc, CancellationToken cancellationToken = default) where TActor : IRuntimeActor
// Declare the lambda return type explicitly, as otherwise the compiler will infer 'ValueTask<TActor>'
// and recurse into the same call, causing a stack overflow.
=> this.RegisterActorFactoryAsync(type, async ValueTask<IRuntimeActor> (actorId, runtime) => await factoryFunc(actorId, runtime).ConfigureAwait(false), cancellationToken);
/// <inheritdoc/>
public async ValueTask<ActorType> RegisterActorFactoryAsync(ActorType type, Func<ActorId, IAgentRuntime, ValueTask<IRuntimeActor>> factoryFunc, CancellationToken cancellationToken = default)
{
if (this._actorFactories.ContainsKey(type))
{
throw new InvalidOperationException($"Actor with type {type} already exists.");
}
this._actorFactories.Add(type, factoryFunc);
return type;
}
/// <inheritdoc/>
public async ValueTask<IdProxyActor?> TryGetActorProxyAsync(ActorId actorId, CancellationToken cancellationToken = default)
{
IdProxyActor proxy = new(this, actorId);
return proxy;
}
private async ValueTask ProcessNextMessageAsync(CancellationToken cancellation = default)
{
if (this._messageDeliveryQueue.TryDequeue(out MessageDelivery? delivery))
{
Interlocked.Decrement(ref this._messageQueueCount);
Debug.WriteLine($"Processing message {delivery.Message.MessageId}...");
await delivery.InvokeAsync(cancellation).ConfigureAwait(false);
}
}
private async Task RunAsync(CancellationToken cancellation)
{
ConcurrentDictionary<Guid, Task> pendingTasks = [];
while (!cancellation.IsCancellationRequested && this._shouldContinue())
{
// Get a unique task id.
Guid taskId = Guid.NewGuid();
// There is potentially a race condition here, but even if we leak a Task, we will
// still catch it on the Finish() pass.
ValueTask processTask = this.ProcessNextMessageAsync(cancellation);
await Task.Yield();
if (!processTask.IsCompleted)
{
pendingTasks.TryAdd(taskId, processTask.AsTask().ContinueWith(t => pendingTasks.TryRemove(taskId, out _), TaskScheduler.Current));
}
}
// The pending task dictionary may contain null values when a race condition is experienced during
// the prior "ContinueWith" call. This could be solved with a ConcurrentDictionary, but locking
// is entirely undesirable in this context.
await Task.WhenAll(pendingTasks.Values.Where(task => task is not null)).ConfigureAwait(false);
await this.FinishAsync(this._finishSource?.Token ?? CancellationToken.None).ConfigureAwait(false);
}
private async ValueTask PublishMessageServicerAsync(MessageEnvelope envelope, CancellationToken deliveryToken)
{
if (!envelope.Topic.HasValue)
{
throw new InvalidOperationException("Message must have a topic to be published.");
}
List<Exception>? exceptions = null;
TopicId topic = envelope.Topic.Value;
foreach (ISubscriptionDefinition subscription in this._subscriptions.Values.Where(subscription => subscription.Matches(topic)))
{
try
{
deliveryToken.ThrowIfCancellationRequested();
ActorId? sender = envelope.Sender;
using CancellationTokenSource combinedSource = CancellationTokenSource.CreateLinkedTokenSource(envelope.Cancellation, deliveryToken);
ActorId actorId = subscription.MapToActor(topic);
if (!this.DeliverToSelf && sender.HasValue && sender == actorId)
{
continue;
}
MessageContext messageContext = new()
{
MessageId = envelope.MessageId,
Sender = sender,
Topic = topic,
IsRpc = false
};
IRuntimeActor actor = await this.EnsureActorAsync(actorId, combinedSource.Token).ConfigureAwait(false);
await actor.OnMessageAsync(envelope.Message, messageContext, combinedSource.Token).ConfigureAwait(false);
}
catch (Exception ex)
{
(exceptions ??= []).Add(ex);
}
}
if (exceptions is not null)
{
throw new AggregateException("One or more exceptions occurred while processing the message.", exceptions);
}
}
private async ValueTask<object?> SendMessageServicerAsync(MessageEnvelope envelope, CancellationToken deliveryToken)
{
if (!envelope.Receiver.HasValue)
{
throw new InvalidOperationException("Message must have a receiver to be sent.");
}
using CancellationTokenSource combinedSource = CancellationTokenSource.CreateLinkedTokenSource(envelope.Cancellation, deliveryToken);
MessageContext messageContext = new()
{
MessageId = envelope.MessageId,
Sender = envelope.Sender,
IsRpc = false
};
ActorId receiver = envelope.Receiver.Value;
IRuntimeActor actor = await this.EnsureActorAsync(receiver, combinedSource.Token).ConfigureAwait(false);
return await actor.OnMessageAsync(envelope.Message, messageContext, combinedSource.Token).ConfigureAwait(false);
}
private async ValueTask<IRuntimeActor> EnsureActorAsync(ActorId actorId, CancellationToken cancellationToken)
{
if (!this._actorInstances.TryGetValue(actorId, out IRuntimeActor? actor))
{
if (!this._actorFactories.TryGetValue(actorId.Type, out Func<ActorId, IAgentRuntime, ValueTask<IRuntimeActor>>? factoryFunc))
{
throw new InvalidOperationException($"Actor with name {actorId.Type} not found.");
}
actor = await factoryFunc(actorId, this).ConfigureAwait(false);
this._actorInstances.Add(actorId, actor);
}
return actor;
}
private async Task FinishAsync(CancellationToken token)
{
foreach (IRuntimeActor actor in this._actorInstances.Values)
{
if (!token.IsCancellationRequested && actor is IAsyncDisposable closeableActor)
{
await closeableActor.DisposeAsync().ConfigureAwait(false);
}
}
if (this._shutdownSource is { } shutdownSource)
{
this._shutdownSource = null;
shutdownSource.Dispose();
}
if (this._finishSource is { } finishSource)
{
this._finishSource = null;
finishSource.Dispose();
}
}
#pragma warning disable CA1822 // Mark members as static
private ValueTask<T> ExecuteTracedAsync<T>(Func<ValueTask<T>> func)
{
// TODO: Bind tracing
return func();
}
private ValueTask ExecuteTracedAsync(Func<ValueTask> func)
{
// TODO: Bind tracing
return func();
}
#pragma warning restore CA1822 // Mark members as static
[JsonSerializable(typeof(Dictionary<string, JsonElement>))]
private sealed partial class InProcessRuntimeContext : JsonSerializerContext;
}
@@ -1,16 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Threading;
using System.Threading.Tasks;
namespace Microsoft.Extensions.AI.Agents.Runtime.InProcess;
internal sealed class MessageDelivery(MessageEnvelope message, Func<MessageEnvelope, CancellationToken, ValueTask> servicer, Task<object?> resultTask)
{
public MessageEnvelope Message { get; } = message;
public Func<MessageEnvelope, CancellationToken, ValueTask> Servicer { get; } = servicer;
public Task<object?> ResultTask { get; } = resultTask;
public ValueTask InvokeAsync(CancellationToken cancellation) => this.Servicer(this.Message, cancellation);
}
@@ -1,65 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Threading;
using System.Threading.Tasks;
namespace Microsoft.Extensions.AI.Agents.Runtime.InProcess;
internal sealed class MessageEnvelope(object message, string? messageId = null, CancellationToken cancellation = default)
{
public object Message { get; } = message;
public string MessageId { get; } = messageId ?? Guid.NewGuid().ToString();
public TopicId? Topic { get; private set; }
public ActorId? Sender { get; private set; }
public ActorId? Receiver { get; private set; }
public CancellationToken Cancellation { get; } = cancellation;
public MessageEnvelope WithSender(ActorId? sender)
{
this.Sender = sender;
return this;
}
public MessageDelivery ForSend(ActorId receiver, Func<MessageEnvelope, CancellationToken, ValueTask<object?>> servicer)
{
this.Receiver = receiver;
TaskCompletionSource<object?> tcs = new();
return new MessageDelivery(this, async (MessageEnvelope envelope, CancellationToken cancellation) =>
{
try
{
object? result = await servicer(envelope, cancellation).ConfigureAwait(false);
tcs.SetResult(result);
}
catch (OperationCanceledException exception)
{
tcs.TrySetCanceled(exception.CancellationToken);
}
catch (Exception exception)
{
tcs.SetException(exception);
}
}, tcs.Task);
}
public MessageDelivery ForPublish(TopicId topic, Func<MessageEnvelope, CancellationToken, ValueTask> servicer)
{
this.Topic = topic;
TaskCompletionSource<object?> tcs = new();
return new MessageDelivery(this, async (envelope, cancellation) =>
{
try
{
await servicer(envelope, cancellation).ConfigureAwait(false);
tcs.SetResult(null);
}
catch (Exception ex)
{
tcs.SetException(ex);
}
}, tcs.Task);
}
}
@@ -1,27 +0,0 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>$(ProjectsTargetFrameworks)</TargetFrameworks>
<TargetFrameworks Condition="'$(Configuration)' == 'Debug'">$(ProjectsDebugTargetFrameworks)</TargetFrameworks>
<VersionSuffix>alpha</VersionSuffix>
</PropertyGroup>
<PropertyGroup>
<InjectDiagnosticAttributesOnLegacy>true</InjectDiagnosticAttributesOnLegacy>
</PropertyGroup>
<ItemGroup>
<InternalsVisibleTo Include="Microsoft.Extensions.AI.Agents.Runtime.InProcess.UnitTests" />
</ItemGroup>
<Import Project="$(RepoRoot)/dotnet/nuget/nuget-package.props" />
<ItemGroup Condition="'$(TargetFrameworkIdentifier)' != '.NETCoreApp'">
<PackageReference Include="System.Threading.Tasks.Extensions" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\Microsoft.Extensions.AI.Agents.Runtime.Abstractions\Microsoft.Extensions.AI.Agents.Runtime.Abstractions.csproj" />
</ItemGroup>
</Project>
@@ -16,11 +16,6 @@ namespace Microsoft.Shared.SampleUtilities;
/// </summary>
public abstract class OrchestrationSample : BaseSample
{
/// <summary>
/// This constant defines the timeout duration for result retrieval, measured in seconds.
/// </summary>
protected const int ResultTimeoutInSeconds = 30;
/// <summary>
/// Creates a new <see cref="ChatClientAgent"/> instance using the specified instructions, description, name, and functions.
/// </summary>
@@ -136,28 +131,23 @@ public abstract class OrchestrationSample : BaseSample
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
public ValueTask ResponseCallback(IEnumerable<ChatMessage> response)
{
WriteStreamedResponse(this.StreamedResponses);
this.StreamedResponses.Clear();
this.History.AddRange(response);
WriteResponse(response);
return new ValueTask();
return default;
}
/// <summary>
/// Callback to handle a streamed agent run response update, adding it to the list and writing output if final.
/// </summary>
/// <param name="streamedResponse">The <see cref="AgentRunResponseUpdate"/> to process.</param>
/// <param name="isFinal">Indicates whether this is the final update in the stream.</param>
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
public ValueTask StreamingResultCallback(AgentRunResponseUpdate streamedResponse, bool isFinal)
public ValueTask StreamingResultCallback(AgentRunResponseUpdate streamedResponse)
{
this.StreamedResponses.Add(streamedResponse);
if (isFinal)
{
WriteStreamedResponse(this.StreamedResponses);
this.StreamedResponses.Clear();
}
return new ValueTask();
return default;
}
}
@@ -1,7 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using Microsoft.Agents.Orchestration.GroupChat;
namespace Microsoft.Agents.Orchestration.UnitTest;
@@ -1,10 +1,7 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Threading.Tasks;
using Microsoft.Agents.Orchestration.Concurrent;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime.InProcess;
namespace Microsoft.Agents.Orchestration.UnitTest;
@@ -17,11 +14,10 @@ public class ConcurrentOrchestrationTests
public async Task ConcurrentOrchestrationWithSingleAgentAsync()
{
// Arrange
await using InProcessRuntime runtime = new();
MockAgent mockAgent1 = MockAgent.CreateWithResponse(1, "xyz");
// Act: Create and execute the orchestration
string[] response = await ExecuteOrchestrationAsync(runtime, mockAgent1);
string[] response = await ExecuteOrchestrationAsync(mockAgent1);
// Assert
Assert.Equal(1, mockAgent1.InvokeCount);
@@ -32,14 +28,12 @@ public class ConcurrentOrchestrationTests
public async Task ConcurrentOrchestrationWithMultipleAgentsAsync()
{
// Arrange
await using InProcessRuntime runtime = new();
MockAgent mockAgent1 = MockAgent.CreateWithResponse(1, "abc");
MockAgent mockAgent2 = MockAgent.CreateWithResponse(2, "xyz");
MockAgent mockAgent3 = MockAgent.CreateWithResponse(3, "lmn");
// Act: Create and execute the orchestration
string[] response = await ExecuteOrchestrationAsync(runtime, mockAgent1, mockAgent2, mockAgent3);
string[] response = await ExecuteOrchestrationAsync(mockAgent1, mockAgent2, mockAgent3);
// Assert
Assert.Equal(1, mockAgent1.InvokeCount);
@@ -50,24 +44,18 @@ public class ConcurrentOrchestrationTests
Assert.Contains("abc", response);
}
private static async Task<string[]> ExecuteOrchestrationAsync(InProcessRuntime runtime, params Agent[] mockAgents)
private static async Task<string[]> ExecuteOrchestrationAsync(params Agent[] mockAgents)
{
// Act
await runtime.StartAsync();
ConcurrentOrchestration orchestration = new(mockAgents);
const string InitialInput = "123";
OrchestrationResult<string[]> result = await orchestration.InvokeAsync(InitialInput, runtime);
OrchestrationResult<string[]> result = await orchestration.InvokeAsync(InitialInput);
// Assert
Assert.NotNull(result);
// Act
string[] response = await result.GetValueAsync(TimeSpan.FromSeconds(20));
await runtime.RunUntilIdleAsync();
return response;
return await result.Task;
}
}
@@ -5,7 +5,6 @@ using System.Collections.Generic;
using System.Linq;
using System.Text.Json;
using System.Threading.Tasks;
using Microsoft.Agents.Orchestration.Transforms;
using Microsoft.Extensions.AI;
namespace Microsoft.Agents.Orchestration.UnitTest;
@@ -1,10 +1,7 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Threading.Tasks;
using Microsoft.Agents.Orchestration.GroupChat;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime.InProcess;
namespace Microsoft.Agents.Orchestration.UnitTest;
@@ -17,11 +14,10 @@ public class GroupChatOrchestrationTests
public async Task GroupChatOrchestrationWithSingleAgentAsync()
{
// Arrange
await using InProcessRuntime runtime = new();
MockAgent mockAgent1 = MockAgent.CreateWithResponse(2, "xyz");
// Act: Create and execute the orchestration
string response = await ExecuteOrchestrationAsync(runtime, mockAgent1);
string response = await ExecuteOrchestrationAsync(mockAgent1);
// Assert
Assert.Equal(1, mockAgent1.InvokeCount);
@@ -32,14 +28,12 @@ public class GroupChatOrchestrationTests
public async Task GroupChatOrchestrationWithMultipleAgentsAsync()
{
// Arrange
await using InProcessRuntime runtime = new();
MockAgent mockAgent1 = MockAgent.CreateWithResponse(1, "abc");
MockAgent mockAgent2 = MockAgent.CreateWithResponse(2, "xyz");
MockAgent mockAgent3 = MockAgent.CreateWithResponse(3, "lmn");
// Act: Create and execute the orchestration
string response = await ExecuteOrchestrationAsync(runtime, mockAgent1, mockAgent2, mockAgent3);
string response = await ExecuteOrchestrationAsync(mockAgent1, mockAgent2, mockAgent3);
// Assert
Assert.Equal(1, mockAgent1.InvokeCount);
@@ -48,24 +42,18 @@ public class GroupChatOrchestrationTests
Assert.Equal("lmn", response);
}
private static async Task<string> ExecuteOrchestrationAsync(InProcessRuntime runtime, params Agent[] mockAgents)
private static async Task<string> ExecuteOrchestrationAsync(params Agent[] mockAgents)
{
// Act
await runtime.StartAsync();
GroupChatOrchestration orchestration = new(new RoundRobinGroupChatManager() { MaximumInvocationCount = mockAgents.Length }, mockAgents);
const string InitialInput = "123";
OrchestrationResult<string> result = await orchestration.InvokeAsync(InitialInput, runtime);
OrchestrationResult<string> result = await orchestration.InvokeAsync(InitialInput);
// Assert
Assert.NotNull(result);
// Act
string response = await result.GetValueAsync(TimeSpan.FromSeconds(20));
await runtime.RunUntilIdleAsync();
return response;
return await result.Task;
}
}
@@ -7,10 +7,8 @@ using System.Collections.Generic;
using System.Net.Http;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Agents.Orchestration.Handoff;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime.InProcess;
using OpenAI;
namespace Microsoft.Agents.Orchestration.UnitTest;
@@ -51,7 +49,7 @@ public sealed class HandoffOrchestrationTests : IDisposable
Responses.Message("Final response"));
// Act: Create and execute the orchestration
string response = await ExecuteOrchestrationAsync(OrchestrationHandoffs.StartWith(mockAgent1), mockAgent1);
string response = await ExecuteOrchestrationAsync(OrchestrationHandoffs.StartWith(mockAgent1));
// Assert
Assert.Equal("Final response", response);
@@ -81,35 +79,26 @@ public sealed class HandoffOrchestrationTests : IDisposable
string response = await ExecuteOrchestrationAsync(
OrchestrationHandoffs
.StartWith(mockAgent1)
.Add(mockAgent1, mockAgent2, mockAgent3),
mockAgent1,
mockAgent2,
mockAgent3);
.Add(mockAgent1, mockAgent2, mockAgent3));
// Assert
Assert.Equal("Final response", response);
}
private static async Task<string> ExecuteOrchestrationAsync(OrchestrationHandoffs handoffs, params Agent[] mockAgents)
private static async Task<string> ExecuteOrchestrationAsync(OrchestrationHandoffs handoffs)
{
// Arrange
await using InProcessRuntime runtime = new();
await runtime.StartAsync();
HandoffOrchestration orchestration = new(handoffs, mockAgents);
HandoffOrchestration orchestration = new(handoffs);
// Act
const string InitialInput = "123";
OrchestrationResult<string> result = await orchestration.InvokeAsync(InitialInput, runtime);
OrchestrationResult<string> result = await orchestration.InvokeAsync(InitialInput);
// Assert
Assert.NotNull(result);
// Act
string response = await result.GetValueAsync(TimeSpan.FromSeconds(10));
await runtime.RunUntilIdleAsync();
return response;
return await result.Task;
}
private ChatClientAgent CreateMockAgent(string name, string description, params string[] responses)
@@ -1,7 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using Microsoft.Agents.Orchestration.Handoff;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents;
using Moq;
@@ -1,6 +1,5 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.AI.Agents.Runtime;
@@ -11,7 +10,7 @@ namespace Microsoft.Agents.Orchestration.UnitTest;
public class OrchestrationResultTests
{
[Fact]
public void ConstructorInitializesPropertiesCorrectly()
public async Task ConstructorInitializesPropertiesCorrectlyAsync()
{
// Arrange
OrchestrationContext context = new("TestOrchestration", new TopicId("testTopic"), null, null, NullLoggerFactory.Instance, CancellationToken.None);
@@ -19,7 +18,7 @@ public class OrchestrationResultTests
// Act
using CancellationTokenSource cancelSource = new();
using OrchestrationResult<string> result = new(context, tcs, cancelSource, NullLogger.Instance);
await using OrchestrationResult<string> result = new(context, tcs, cancelSource, NullLogger.Instance);
// Assert
Assert.Equal("TestOrchestration", result.Orchestration);
@@ -33,51 +32,17 @@ public class OrchestrationResultTests
OrchestrationContext context = new("TestOrchestration", new TopicId("testTopic"), null, null, NullLoggerFactory.Instance, CancellationToken.None);
TaskCompletionSource<string> tcs = new();
using CancellationTokenSource cancelSource = new();
using OrchestrationResult<string> result = new(context, tcs, cancelSource, NullLogger.Instance);
await using OrchestrationResult<string> result = new(context, tcs, cancelSource, NullLogger.Instance);
string expectedValue = "Result value";
// Act
tcs.SetResult(expectedValue);
string actualValue = await result.GetValueAsync();
string actualValue = await result.Task;
// Assert
Assert.Equal(expectedValue, actualValue);
}
[Fact]
public async Task GetValueAsyncWithTimeoutReturnsCompletedValueWhenTaskCompletesWithinTimeoutAsync()
{
// Arrange
OrchestrationContext context = new("TestOrchestration", new TopicId("testTopic"), null, null, NullLoggerFactory.Instance, CancellationToken.None);
TaskCompletionSource<string> tcs = new();
using CancellationTokenSource cancelSource = new();
using OrchestrationResult<string> result = new(context, tcs, cancelSource, NullLogger.Instance);
string expectedValue = "Result value";
TimeSpan timeout = TimeSpan.FromSeconds(1);
// Act
tcs.SetResult(expectedValue);
string actualValue = await result.GetValueAsync(timeout);
// Assert
Assert.Equal(expectedValue, actualValue);
}
[Fact]
public async Task GetValueAsyncWithTimeoutThrowsTimeoutExceptionWhenTaskDoesNotCompleteWithinTimeoutAsync()
{
// Arrange
OrchestrationContext context = new("TestOrchestration", new TopicId("testTopic"), null, null, NullLoggerFactory.Instance, CancellationToken.None);
TaskCompletionSource<string> tcs = new();
using CancellationTokenSource cancelSource = new();
using OrchestrationResult<string> result = new(context, tcs, cancelSource, NullLogger.Instance);
TimeSpan timeout = TimeSpan.FromMilliseconds(50);
// Act & Assert
TimeoutException exception = await Assert.ThrowsAsync<TimeoutException>(() => result.GetValueAsync(timeout).AsTask());
Assert.Contains("Orchestration did not complete within the allowed duration", exception.Message);
}
[Fact]
public async Task GetValueAsyncReturnsCompletedValueWhenCompletionIsDelayedAsync()
{
@@ -85,7 +50,7 @@ public class OrchestrationResultTests
OrchestrationContext context = new("TestOrchestration", new TopicId("testTopic"), null, null, NullLoggerFactory.Instance, CancellationToken.None);
TaskCompletionSource<int> tcs = new();
using CancellationTokenSource cancelSource = new();
using OrchestrationResult<int> result = new(context, tcs, cancelSource, NullLogger.Instance);
await using OrchestrationResult<int> result = new(context, tcs, cancelSource, NullLogger.Instance);
int expectedValue = 42;
// Act
@@ -96,7 +61,7 @@ public class OrchestrationResultTests
tcs.SetResult(expectedValue);
});
int actualValue = await result.GetValueAsync();
int actualValue = await result.Task;
// Assert
Assert.Equal(expectedValue, actualValue);
@@ -1,10 +1,7 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Threading.Tasks;
using Microsoft.Agents.Orchestration.Sequential;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Extensions.AI.Agents.Runtime.InProcess;
namespace Microsoft.Agents.Orchestration.UnitTest;
@@ -17,11 +14,10 @@ public class SequentialOrchestrationTests
public async Task SequentialOrchestrationWithSingleAgentAsync()
{
// Arrange
await using InProcessRuntime runtime = new();
MockAgent mockAgent1 = MockAgent.CreateWithResponse(2, "xyz");
// Act: Create and execute the orchestration
string response = await ExecuteOrchestrationAsync(runtime, mockAgent1);
string response = await ExecuteOrchestrationAsync(mockAgent1);
// Assert
Assert.Equal(1, mockAgent1.InvokeCount);
@@ -32,14 +28,12 @@ public class SequentialOrchestrationTests
public async Task SequentialOrchestrationWithMultipleAgentsAsync()
{
// Arrange
await using InProcessRuntime runtime = new();
MockAgent mockAgent1 = MockAgent.CreateWithResponse(1, "abc");
MockAgent mockAgent2 = MockAgent.CreateWithResponse(2, "xyz");
MockAgent mockAgent3 = MockAgent.CreateWithResponse(3, "lmn");
// Act: Create and execute the orchestration
string response = await ExecuteOrchestrationAsync(runtime, mockAgent1, mockAgent2, mockAgent3);
string response = await ExecuteOrchestrationAsync(mockAgent1, mockAgent2, mockAgent3);
// Assert
Assert.Equal(1, mockAgent1.InvokeCount);
@@ -48,24 +42,18 @@ public class SequentialOrchestrationTests
Assert.Equal("lmn", response);
}
private static async Task<string> ExecuteOrchestrationAsync(InProcessRuntime runtime, params Agent[] mockAgents)
private static async Task<string> ExecuteOrchestrationAsync(params Agent[] mockAgents)
{
// Act
await runtime.StartAsync();
SequentialOrchestration orchestration = new(mockAgents);
const string InitialInput = "123";
OrchestrationResult<string> result = await orchestration.InvokeAsync(InitialInput, runtime);
OrchestrationResult<string> result = await orchestration.InvokeAsync(InitialInput);
// Assert
Assert.NotNull(result);
// Act
string response = await result.GetValueAsync(TimeSpan.FromSeconds(20));
await runtime.RunUntilIdleAsync();
return response;
return await result.Task;
}
}
@@ -18,25 +18,20 @@ public class InProcessRuntimeTests()
await using InProcessRuntime runtime = new();
// Assert
Assert.False(runtime.DeliverToSelf);
Assert.Equal(0, runtime._messageQueueCount);
Assert.Equal(0, runtime.MessageCountForTesting);
// Act
await runtime.StopAsync(); // Already stopped
await runtime.RunUntilIdleAsync(); // Never throws
await runtime.StartAsync();
runtime.Start();
// Assert
// Invalid to start runtime that is already started
await Assert.ThrowsAsync<InvalidOperationException>(() => runtime.StartAsync());
Assert.Equal(0, runtime._messageQueueCount);
Assert.Throws<InvalidOperationException>(runtime.Start);
Assert.Equal(0, runtime.MessageCountForTesting);
// Act
await runtime.StopAsync();
await runtime.DisposeAsync();
// Assert
Assert.Equal(0, runtime._messageQueueCount);
Assert.Equal(0, runtime.MessageCountForTesting);
}
[Fact]
@@ -194,26 +189,21 @@ public class InProcessRuntimeTests()
Assert.Empty(agent.ReceivedMessages);
// Act: Send message
await runtime.StartAsync();
runtime.Start();
await runtime.SendMessageAsync("TestMessage", agent.Id);
await runtime.RunUntilIdleAsync();
await runtime.DisposeAsync();
// Assert
Assert.Equal(0, runtime._messageQueueCount);
Assert.Equal(0, runtime.MessageCountForTesting);
Assert.Single(agent.ReceivedMessages);
}
// Agent will not deliver to self will success when runtime.DeliverToSelf is false (default)
[Theory]
[InlineData(false, 0)]
[InlineData(true, 1)]
public async Task RuntimeAgentPublishToSelfTestAsync(bool selfPublish, int receiveCount)
// Agent will not deliver to self
[Fact]
public async Task RuntimeAgentPublishToSelfTestAsync()
{
// Arrange
await using InProcessRuntime runtime = new()
{
DeliverToSelf = selfPublish
};
await using InProcessRuntime runtime = new();
MockAgent? agent = null;
await runtime.RegisterActorFactoryAsync(new("MyAgent"), async (id, runtime) =>
@@ -238,12 +228,12 @@ public class InProcessRuntimeTests()
await runtime.AddSubscriptionAsync(new TestSubscription(TopicType, agentId.Type));
// Act
await runtime.StartAsync();
runtime.Start();
await runtime.PublishMessageAsync("SelfMessage", new TopicId(TopicType), sender: agentId);
await runtime.RunUntilIdleAsync();
await runtime.DisposeAsync();
// Assert
Assert.Equal(receiveCount, agent.ReceivedMessages.Count);
Assert.Empty(agent.ReceivedMessages);
}
[Fact]
@@ -263,9 +253,9 @@ public class InProcessRuntimeTests()
const string TopicType = "TestTopic";
await runtime.AddSubscriptionAsync(new TestSubscription(TopicType, agentId.Type));
await runtime.StartAsync();
runtime.Start();
await runtime.PublishMessageAsync("test", new TopicId(TopicType));
await runtime.RunUntilIdleAsync();
await runtime.DisposeAsync();
// Act: Save the state
JsonElement savedState = await runtime.SaveStateAsync();
@@ -286,8 +276,7 @@ public class InProcessRuntimeTests()
// Act: Start new runtime and restore the state
agent = null;
await using InProcessRuntime newRuntime = new();
await newRuntime.StartAsync();
await using InProcessRuntime newRuntime = InProcessRuntime.StartNew();
await newRuntime.RegisterActorFactoryAsync(new("MyAgent"), async (id, runtime) =>
{
agent = new MockAgent(id, runtime, "another agent");
@@ -56,10 +56,7 @@ public sealed class ReceiverAgent : TestAgent
public ReceiverAgent(ActorId id, IAgentRuntime runtime, string description) : base(id, runtime, description)
{
this.RegisterMessageHandler<BasicMessage>(async (item, messageContext, cancellationToken) =>
{
this.Messages.Add(item);
});
this.RegisterMessageHandler<BasicMessage>((item, messageContext) => this.Messages.Add(item));
}
}
@@ -78,10 +75,10 @@ public sealed class CancelAgent : TestAgent
{
public CancelAgent(ActorId id, IAgentRuntime runtime, string description) : base(id, runtime, description)
{
this.RegisterMessageHandler<BasicMessage>(async (item, messageContext, cancellationToken) =>
this.RegisterMessageHandler<BasicMessage>((item, messageContext) =>
{
CancellationToken cancelledToken = new(canceled: true);
cancelledToken.ThrowIfCancellationRequested();
CancellationToken canceledToken = new(canceled: true);
canceledToken.ThrowIfCancellationRequested();
});
}
}
@@ -90,7 +87,7 @@ public sealed class ErrorAgent : TestAgent
{
public ErrorAgent(ActorId id, IAgentRuntime runtime, string description) : base(id, runtime, description)
{
this.RegisterMessageHandler<BasicMessage>(async (item, messageContext, cancellationToken) =>
this.RegisterMessageHandler<BasicMessage>((item, messageContext) =>
{
this.DidThrow = true;
throw new TestException();
@@ -155,20 +152,20 @@ public sealed class MessagingTestFixture
{
messageId ??= Guid.NewGuid().ToString();
await this.Runtime.StartAsync();
this.Runtime.Start();
await this.Runtime.PublishMessageAsync(message, sendTarget, messageId: messageId);
await this.Runtime.RunUntilIdleAsync();
await this.Runtime.DisposeAsync();
}
public async ValueTask<object?> RunSendTestAsync(ActorId sendTarget, object message, string? messageId = null)
{
messageId ??= Guid.NewGuid().ToString();
await this.Runtime.StartAsync();
this.Runtime.Start();
object? result = await this.Runtime.SendMessageAsync(message, sendTarget, messageId: messageId);
await this.Runtime.RunUntilIdleAsync();
await this.Runtime.DisposeAsync();
return result;
}
@@ -26,13 +26,12 @@ public sealed class MockAgent : TestAgent
{
public MockAgent(ActorId id, IAgentRuntime runtime, string description) : base(id, runtime, description)
{
this.RegisterMessageHandler<string>(this.HandleAsync);
this.RegisterMessageHandler<string>(this.Handle);
}
public ValueTask HandleAsync(string item, MessageContext messageContext, CancellationToken cancellationToken)
public void Handle(string item, MessageContext messageContext)
{
this.ReceivedMessages.Add(item);
return default;
}
public override async ValueTask<JsonElement> SaveStateAsync(CancellationToken cancellationToken = default)
@@ -1,7 +1,6 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Diagnostics.CodeAnalysis;
namespace Microsoft.Extensions.AI.Agents.Runtime.InProcess.Tests;
@@ -23,7 +22,7 @@ public class TestSubscription(string topicType, ActorType agentType, string? id
public bool Equals(ISubscriptionDefinition? other) => this.Id == other?.Id;
public override bool Equals([NotNullWhen(true)] object? obj) => obj is TestSubscription other && other.Equals(this);
public override bool Equals(object? obj) => obj is TestSubscription other && other.Equals(this);
public override int GetHashCode() => this.Id.GetHashCode();
@@ -1,101 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Threading;
using System.Threading.Tasks;
namespace Microsoft.Extensions.AI.Agents.Runtime.InProcess.Tests;
public class MessageEnvelopeTests
{
[Fact]
public void ConstructAllParametersTest()
{
// Arrange
object message = new { Content = "Test message" };
const string MessageId = "testid";
CancellationToken cancellation = new();
// Act
MessageEnvelope envelope = new(message, MessageId, cancellation);
// Assert
Assert.Same(message, envelope.Message);
Assert.Equal(MessageId, envelope.MessageId);
Assert.Equal(cancellation, envelope.Cancellation);
Assert.Null(envelope.Sender);
Assert.Null(envelope.Receiver);
Assert.Null(envelope.Topic);
}
[Fact]
public void ConstructOnlyRequiredParametersTest()
{
// Arrange & Act
MessageEnvelope envelope = new("test");
// Assert
Assert.NotNull(envelope.MessageId);
Assert.NotEmpty(envelope.MessageId);
// Verify it's a valid GUID
Assert.True(Guid.TryParse(envelope.MessageId, out _));
}
[Fact]
public void WithSenderTest()
{
// Arrange
MessageEnvelope envelope = new("test");
ActorId sender = new("testtype", "testkey");
// Act
MessageEnvelope result = envelope.WithSender(sender);
// Assert
Assert.Same(envelope, result);
Assert.Equal(sender, envelope.Sender);
}
[Fact]
public async Task ForSendTestAsync()
{
// Arrange
MessageEnvelope envelope = new("test");
ActorId receiver = new("receivertype", "receiverkey");
object expectedResult = new { Response = "Success" };
ValueTask<object?> servicer(MessageEnvelope env, CancellationToken ct) => new(expectedResult);
// Act
MessageDelivery delivery = envelope.ForSend(receiver, servicer);
// Assert
Assert.NotNull(delivery);
Assert.Same(envelope, delivery.Message);
Assert.Equal(receiver, envelope.Receiver);
// Invoke the servicer to verify result sink works
await delivery.InvokeAsync(CancellationToken.None);
Assert.True(delivery.ResultTask.IsCompleted);
object? result = await delivery.ResultTask;
Assert.Same(expectedResult, result);
}
[Fact]
public void ForPublishTest()
{
// Arrange
MessageEnvelope envelope = new("test");
TopicId topic = new("testtopic");
static ValueTask servicer(MessageEnvelope env, CancellationToken ct) => default;
// Act
MessageDelivery delivery = envelope.ForPublish(topic, servicer);
// Assert
Assert.NotNull(delivery);
Assert.Same(envelope, delivery.Message);
Assert.Equal(topic, envelope.Topic);
}
}
@@ -1,12 +0,0 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>$(ProjectsTargetFrameworks)</TargetFrameworks>
<TargetFrameworks Condition="'$(Configuration)' == 'Debug'">$(ProjectsDebugTargetFrameworks)</TargetFrameworks>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\src\Microsoft.Extensions.AI.Agents.Runtime.InProcess\Microsoft.Extensions.AI.Agents.Runtime.InProcess.csproj" />
</ItemGroup>
</Project>