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

Overhaul orchestration library with new approach (#199)

Browse Source
This commit is contained in:
Stephen Toub
2025-07-23 16:29:37 +00:00
committed by GitHub
parent 27f7af2160
commit 5472d6e996
104 changed files with 1212 additions and 5084 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/ActorJsonContext.cs
+18 -18
View File
@@ -6,36 +6,36 @@ using System.Text.Json.Serialization;
namespace Microsoft.Extensions.AI.Agents.Runtime;
/// <summary>
/// Source-generated JSON type information for use by all Actor abstractions.
/// Source-generated JSON type information for use by all agent runtime abstractions.
/// </summary>
[JsonSourceGenerationOptions(
JsonSerializerDefaults.Web,
UseStringEnumConverter = true,
DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull,
WriteIndented = false)]
[JsonSerializable(typeof(ActorId))]
[JsonSerializable(typeof(ActorMessage))]
[JsonSerializable(typeof(ActorRequestMessage))]
[JsonSerializable(typeof(ActorResponseMessage))]
[JsonSerializable(typeof(ActorWriteOperation))]
[JsonSerializable(typeof(SetValueOperation))]
[JsonSerializable(typeof(RemoveKeyOperation))]
[JsonSerializable(typeof(SendRequestOperation))]
[JsonSerializable(typeof(UpdateRequestOperation))]
[JsonSerializable(typeof(ActorReadOperation))]
[JsonSerializable(typeof(ListKeysOperation))]
[JsonSerializable(typeof(GetValueOperation))]
[JsonSerializable(typeof(ActorReadOperationBatch))]
[JsonSerializable(typeof(ActorReadResult))]
[JsonSerializable(typeof(ListKeysResult))]
[JsonSerializable(typeof(GetValueResult))]
[JsonSerializable(typeof(ActorRequest))]
[JsonSerializable(typeof(ActorRequestMessage))]
[JsonSerializable(typeof(ActorRequestUpdate))]
[JsonSerializable(typeof(ActorResponse))]
[JsonSerializable(typeof(ActorId))]
[JsonSerializable(typeof(RequestStatus))]
[JsonSerializable(typeof(ActorWriteOperationBatch))]
[JsonSerializable(typeof(ActorReadOperationBatch))]
[JsonSerializable(typeof(ReadResponse))]
[JsonSerializable(typeof(WriteResponse))]
[JsonSerializable(typeof(ActorResponseMessage))]
[JsonSerializable(typeof(ActorType))]
[JsonSerializable(typeof(ActorWriteOperation))]
[JsonSerializable(typeof(ActorWriteOperationBatch))]
[JsonSerializable(typeof(GetValueOperation))]
[JsonSerializable(typeof(GetValueResult))]
[JsonSerializable(typeof(JsonElement))]
[JsonSerializable(typeof(ListKeysOperation))]
[JsonSerializable(typeof(ListKeysResult))]
[JsonSerializable(typeof(ReadResponse))]
[JsonSerializable(typeof(RemoveKeyOperation))]
[JsonSerializable(typeof(RequestStatus))]
[JsonSerializable(typeof(SendRequestOperation))]
[JsonSerializable(typeof(SetValueOperation))]
[JsonSerializable(typeof(UpdateRequestOperation))]
[JsonSerializable(typeof(WriteResponse))]
internal sealed partial class ActorJsonContext : JsonSerializerContext;
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/ActorMetadata.cs
-67
View File
@@ -1,67 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
namespace Microsoft.Extensions.AI.Agents.Runtime;
/// <summary>
/// Represents metadata associated with an actor, including its type, unique key, and description.
/// </summary>
public readonly struct ActorMetadata : IEquatable<ActorMetadata>
{
/// <summary>
/// Initializes a new instance of the <see cref="ActorMetadata"/> class with the specified type, key, and description.
/// </summary>
/// <param name="type">The type of the actor.</param>
/// <param name="key">The unique key associated with the actor.</param>
/// <param name="description">A brief description of the actor.</param>
public ActorMetadata(ActorType type, string key, string? description = null)
{
if (!ActorId.IsValidKey(key))
{
throw new ArgumentException("Invalid actor key.", nameof(key));
}
this.Type = type;
this.Key = key;
this.Description = description;
}
/// <summary>
/// Gets an identifier that associates an actor with a specific factory function.
/// </summary>
public ActorType Type { get; }
/// <summary>
/// A unique key identifying the actor instance.
/// Strings may only be composed of alphanumeric letters (a-z, 0-9), or underscores (_).
/// </summary>
public string Key { get; }
/// <summary>
/// A brief description of the actor's purpose or functionality.
/// </summary>
public string? Description { get; }
/// <inheritdoc/>
public override readonly bool Equals(object? obj) =>
obj is ActorMetadata actorMetadata && this.Equals(actorMetadata);
/// <inheritdoc/>
public readonly bool Equals(ActorMetadata other) =>
this.Type == other.Type &&
this.Key == other.Key &&
this.Description == other.Description;
/// <inheritdoc/>
public override readonly int GetHashCode() =>
HashCode.Combine(this.Type, this.Key, this.Description);
/// <inheritdoc/>
public static bool operator ==(ActorMetadata left, ActorMetadata right) =>
left.Equals(right);
/// <inheritdoc/>
public static bool operator !=(ActorMetadata left, ActorMetadata right) =>
!(left == right);
}
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/ActorReadOperationBatch.cs
+1 -1
View File
@@ -9,7 +9,7 @@ namespace Microsoft.Extensions.AI.Agents.Runtime;
/// Represents a batch of read operations to be performed on an actor.
/// </summary>
/// <param name="operations">The collection of read operations to perform.</param>
public class ActorReadOperationBatch(IReadOnlyList<ActorReadOperation> operations)
public sealed class ActorReadOperationBatch(IReadOnlyList<ActorReadOperation> operations)
{
/// <summary>
/// Gets the collection of read operations to perform.
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/ActorRequest.cs
+1 -1
View File
@@ -8,7 +8,7 @@ namespace Microsoft.Extensions.AI.Agents.Runtime;
/// <summary>
/// Represents a request to be sent to an actor.
/// </summary>
public class ActorRequest(ActorId actorId, string messageId, string method, JsonElement @params)
public sealed class ActorRequest(ActorId actorId, string messageId, string method, JsonElement @params)
{
/// <summary>
/// Gets or sets the identifier of the target actor.
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/ActorRequestUpdate.cs
+1 -1
View File
@@ -10,7 +10,7 @@ namespace Microsoft.Extensions.AI.Agents.Runtime;
/// <summary>
/// Represents an update to an actor request's status and data.
/// </summary>
public class ActorRequestUpdate(RequestStatus status, JsonElement data)
public sealed class ActorRequestUpdate(RequestStatus status, JsonElement data)
{
/// <summary>
/// Gets the updated status of the request.
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/ActorResponse.cs
+1 -1
View File
@@ -8,7 +8,7 @@ namespace Microsoft.Extensions.AI.Agents.Runtime;
/// <summary>
/// Represents a response handle for an actor request, providing access to the result and status updates.
/// </summary>
public class ActorResponse
public sealed class ActorResponse
{
/// <summary>
/// Gets the identifier of the actor that is processing the request.
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/ActorWriteOperationBatch.cs
+1 -1
View File
@@ -10,7 +10,7 @@ namespace Microsoft.Extensions.AI.Agents.Runtime;
/// </summary>
/// <param name="eTag">The ETag for optimistic concurrency control.</param>
/// <param name="operations">The collection of write operations to perform.</param>
public class ActorWriteOperationBatch(string eTag, IReadOnlyCollection<ActorWriteOperation> operations)
public sealed class ActorWriteOperationBatch(string eTag, IReadOnlyCollection<ActorWriteOperation> operations)
{
/// <summary>
/// Gets the collection of write operations to perform.
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/AgentRuntimeExtensions.cs
-71
View File
@@ -1,71 +0,0 @@
// 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);
}
}
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/GetValueOperation.cs
+1 -1
View File
@@ -8,7 +8,7 @@ namespace Microsoft.Extensions.AI.Agents.Runtime;
/// Represents a request to read a value from the actor's state by its key.
/// </summary>
/// <param name="key">The key corresponding to the value to read from the actor's state.</param>
public class GetValueOperation(string key) : ActorStateReadOperation
public sealed class GetValueOperation(string key) : ActorStateReadOperation
{
/// <summary>
/// Gets the key corresponding to the value to read from the actor's state.
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/GetValueResult.cs
+1 -1
View File
@@ -9,7 +9,7 @@ namespace Microsoft.Extensions.AI.Agents.Runtime;
/// Represents the result of a get value operation containing the retrieved value.
/// </summary>
/// <param name="value">The value retrieved from the actor's state, or null if not found.</param>
public class GetValueResult(JsonElement? value) : ActorReadResult
public sealed class GetValueResult(JsonElement? value) : ActorReadResult
{
/// <summary>
/// Gets the value retrieved from the actor's state.
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/IAgentRuntime.cs
-110
View File
@@ -1,110 +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;
namespace Microsoft.Extensions.AI.Agents.Runtime;
/// <summary>
/// Defines the runtime environment for actors, managing message sending, subscriptions, actor resolution, and state persistence,
/// all in support of agent-based architectures.
/// </summary>
public interface IAgentRuntime : ISaveState
{
/// <summary>
/// Sends a message to an actor and gets a response.
/// This method should be used to communicate directly with an actor.
/// </summary>
/// <param name="message">The message to send.</param>
/// <param name="recipient">The actor to send the message to.</param>
/// <param name="sender">The actor sending the message. Should be <c>null</c> if sent from an external source.</param>
/// <param name="messageId">A unique identifier for the message. If <c>null</c>, a new ID will be generated.</param>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>A task representing the asynchronous operation, returning the response from the actor.</returns>
ValueTask<object?> SendMessageAsync(object message, ActorId recipient, ActorId? sender = null, string? messageId = null, CancellationToken cancellationToken = default);
/// <summary>
/// Publishes a message to all agents subscribed to the given topic.
/// No responses are expected from publishing.
/// </summary>
/// <param name="message">The message to publish.</param>
/// <param name="topic">The topic to publish the message to.</param>
/// <param name="sender">The actor sending the message. Defaults to <c>null</c>.</param>
/// <param name="messageId">A unique message ID. If <c>null</c>, a new one will be generated.</param>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>A task representing the asynchronous operation.</returns>
ValueTask PublishMessageAsync(object message, TopicId topic, ActorId? sender = null, string? messageId = null, CancellationToken cancellationToken = default);
/// <summary>
/// Retrieves an actor by its unique identifier.
/// </summary>
/// <param name="actorId">The unique identifier of the actor.</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(ActorId actorId, bool lazy = true, CancellationToken cancellationToken = default);
/// <summary>
/// Saves the state of an actor.
/// The result must be JSON serializable.
/// </summary>
/// <param name="actorId">The ID of the actor whose state is being saved.</param>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>A task representing the asynchronous operation, returning a dictionary of the saved state.</returns>
ValueTask<JsonElement> SaveActorStateAsync(ActorId actorId, CancellationToken cancellationToken = default);
/// <summary>
/// Loads the saved state into an actor.
/// </summary>
/// <param name="actorId">The ID of the actor whose state is being restored.</param>
/// <param name="state">The state dictionary to restore.</param>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>A task representing the asynchronous operation.</returns>
ValueTask LoadActorStateAsync(ActorId actorId, JsonElement state, CancellationToken cancellationToken = default);
/// <summary>
/// Retrieves metadata for an actor.
/// </summary>
/// <param name="actorId">The ID of the actor.</param>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>A task representing the asynchronous operation, returning the actor's metadata.</returns>
ValueTask<ActorMetadata> GetActorMetadataAsync(ActorId actorId, CancellationToken cancellationToken = default);
/// <summary>
/// Adds a new subscription for the runtime to handle when processing published messages.
/// </summary>
/// <param name="subscription">The subscription to add.</param>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>A task representing the asynchronous operation.</returns>
ValueTask AddSubscriptionAsync(ISubscriptionDefinition subscription, CancellationToken cancellationToken = default);
/// <summary>
/// Removes a subscription from the runtime.
/// </summary>
/// <param name="subscriptionId">The unique identifier of the subscription to remove.</param>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>A task representing the asynchronous operation.</returns>
/// <exception cref="KeyNotFoundException">Thrown if the subscription does not exist.</exception>
ValueTask RemoveSubscriptionAsync(string subscriptionId, CancellationToken cancellationToken = default);
/// <summary>
/// Registers an actor factory with the runtime, associating it with a specific actor type.
/// The type must be unique.
/// </summary>
/// <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 <see cref="ActorType"/>.</returns>
ValueTask<ActorType> RegisterActorFactoryAsync(ActorType type, Func<ActorId, IAgentRuntime, ValueTask<IRuntimeActor>> factoryFunc, CancellationToken cancellationToken = default);
/// <summary>
/// Attempts to retrieve an <see cref="IdProxyActor"/> for the specified actor.
/// </summary>
/// <param name="actorId">The ID of the actor.</param>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>A task representing the asynchronous operation, returning an <see cref="IdProxyActor"/> if successful.</returns>
ValueTask<IdProxyActor?> TryGetActorProxyAsync(ActorId actorId, CancellationToken cancellationToken = default);
}
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/IRuntimeActor.cs
-37
View File
@@ -1,37 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Threading;
using System.Threading.Tasks;
namespace Microsoft.Extensions.AI.Agents.Runtime;
/// <summary>
/// Represents an actor within the runtime that can process messages, maintain state, and be closed when no longer needed.
/// </summary>
public interface IRuntimeActor : ISaveState
{
/// <summary>
/// Gets the unique identifier of the actor.
/// </summary>
ActorId Id { get; }
/// <summary>
/// Gets metadata associated with the actor.
/// </summary>
ActorMetadata Metadata { get; }
/// <summary>
/// Handles an incoming message for the actor.
/// This should only be called by the runtime, not by other actors.
/// </summary>
/// <param name="message">The received message. The type should match one of the expected subscription types.</param>
/// <param name="messageContext">The context of the message, providing additional metadata.</param>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>
/// 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 canceled.</exception>
ValueTask<object?> OnMessageAsync(object message, MessageContext messageContext, CancellationToken cancellationToken = default);
}
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/ISaveState.cs
-39
View File
@@ -1,39 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
namespace Microsoft.Extensions.AI.Agents.Runtime;
// TODO: Why is this interface needed? It's inherited by IAgentRuntime and IRuntimeActor.
// Is the former needed (does IAgentRuntime need to not only persist every actor but do so via
// this interface)? If not, these methods could be moved to IRuntimeActor.
/// <summary>
/// Defines a contract for saving and loading the state of an object as JSON.
/// </summary>
public interface ISaveState
{
/// <summary>
/// Saves the current state of the object.
/// </summary>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>
/// A task representing the asynchronous operation, returning a dictionary
/// containing the saved state. The structure of the state is implementation-defined
/// but must be JSON serializable.
/// </returns>
ValueTask<JsonElement> SaveStateAsync(CancellationToken cancellationToken = default);
/// <summary>
/// Loads a previously saved state into the object.
/// </summary>
/// <param name="state">
/// A dictionary representing the saved state. The structure of the state
/// is implementation-defined but must be JSON serializable.
/// </param>
/// <param name="cancellationToken">A token to cancel the operation if needed.</param>
/// <returns>A task representing the asynchronous operation.</returns>
ValueTask LoadStateAsync(JsonElement state, CancellationToken cancellationToken = default);
}
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/ISubscriptionDefinition.cs
-51
View File
@@ -1,51 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics.CodeAnalysis;
namespace Microsoft.Extensions.AI.Agents.Runtime;
/// <summary>
/// Defines a subscription that matches topics and maps them to actors.
/// </summary>
public interface ISubscriptionDefinition
{
/// <summary>
/// Gets the unique identifier of the subscription.
/// </summary>
string Id { get; }
/// <summary>
/// Determines whether the specified object is equal to the current subscription.
/// </summary>
/// <param name="obj">The object to compare with the current instance.</param>
/// <returns><c>true</c> if the specified object is equal to this instance; otherwise, <c>false</c>.</returns>
bool Equals([NotNullWhen(true)] object? obj);
/// <summary>
/// Determines whether the specified subscription is equal to the current subscription.
/// </summary>
/// <param name="other">The subscription to compare.</param>
/// <returns><c>true</c> if the subscriptions are equal; otherwise, <c>false</c>.</returns>
bool Equals(ISubscriptionDefinition? other);
/// <summary>
/// Returns a hash code for this subscription.
/// </summary>
/// <returns>A hash code for the subscription.</returns>
int GetHashCode();
/// <summary>
/// Checks if a given <see cref="TopicId"/> matches the subscription.
/// </summary>
/// <param name="topic">The topic to check.</param>
/// <returns><c>true</c> if the topic matches the subscription; otherwise, <c>false</c>.</returns>
bool Matches(TopicId topic);
/// <summary>
/// Maps a <see cref="TopicId"/> to an <see cref="ActorId"/>.
/// Should only be called if <see cref="Matches"/> returns <c>true</c>.
/// </summary>
/// <param name="topic">The topic to map.</param>
/// <returns>The <see cref="ActorId"/> that should handle the topic.</returns>
ActorId MapToActor(TopicId topic);
}
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/IdProxyActor.cs
-56
View File
@@ -1,56 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Shared.Diagnostics;
namespace Microsoft.Extensions.AI.Agents.Runtime;
/// <summary>
/// Provides an actor proxy that allows you to use an <see cref="ActorId"/> in place of its associated <see cref="IRuntimeActor"/>.
/// </summary>
public sealed class IdProxyActor : IRuntimeActor
{
/// <summary>The runtime instance used to interact with actors.</summary>
private readonly IAgentRuntime _runtime;
/// <summary>The metadata for the actor, lazy-loaded.</summary>
private ActorMetadata? _metadata;
/// <summary>
/// Initializes a new instance of the <see cref="IdProxyActor"/> class.
/// </summary>
public IdProxyActor(IAgentRuntime runtime, ActorId actorId)
{
Throw.IfNull(runtime);
this.Id = actorId;
this._runtime = runtime;
}
/// <inheritdoc />
public ActorId Id { get; }
/// <inheritdoc />
public ActorMetadata Metadata =>
this._metadata ??=
#pragma warning disable VSTHRD002 // Avoid problematic synchronous waits
this._runtime.GetActorMetadataAsync(this.Id).AsTask().GetAwaiter().GetResult();
#pragma warning restore VSTHRD002
/// <inheritdoc />
public ValueTask<object?> SendMessageAsync(object message, ActorId sender, string? messageId = null, CancellationToken cancellationToken = default) =>
this._runtime.SendMessageAsync(message, this.Id, sender, messageId, cancellationToken);
/// <inheritdoc />
public ValueTask LoadStateAsync(JsonElement state, CancellationToken cancellationToken = default) =>
this._runtime.LoadActorStateAsync(this.Id, state, cancellationToken);
/// <inheritdoc />
public ValueTask<JsonElement> SaveStateAsync(CancellationToken cancellationToken = default) =>
this._runtime.SaveActorStateAsync(this.Id, cancellationToken);
/// <inheritdoc />
ValueTask<object?> IRuntimeActor.OnMessageAsync(object message, MessageContext messageContext, CancellationToken cancellationToken) =>
new((object?)null);
}
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/InProcessRuntime.cs
-429
View File
@@ -1,429 +0,0 @@
// 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)
{
Throw.IfNull(message);
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)
{
Throw.IfNull(message);
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)
{
Throw.IfNull(subscription);
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)
{
Throw.IfNull(subscriptionId);
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)
{
Throw.IfNull(factoryFunc);
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<Task>? tasks = null;
TopicId topic = message.Topic!.Value;
foreach (KeyValuePair<string, ISubscriptionDefinition> subscription in message.Runtime._subscriptions)
{
if (subscription.Value.Matches(topic))
{
(tasks ??= []).Add(ProcessSubscriptionAsync(message, subscription.Value, topic, cancellationToken));
}
static async Task ProcessSubscriptionAsync(
MessageToProcess message, ISubscriptionDefinition subscription, TopicId topic, CancellationToken cancellationToken)
{
using CancellationTokenSource combinedSource = CancellationTokenSource.CreateLinkedTokenSource(message.Cancellation, cancellationToken);
combinedSource.Token.ThrowIfCancellationRequested();
ActorId actorId = subscription.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);
}
}
}
if (tasks is not null)
{
await Task.WhenAll(tasks).ConfigureAwait(false);
}
// 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);
}
}
}
}
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/KeyValueParserExtensions.cs → dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/KeyValueParser.cs
View File
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/ListKeysOperation.cs
+1 -1
View File
@@ -9,7 +9,7 @@ namespace Microsoft.Extensions.AI.Agents.Runtime;
/// </summary>
/// <param name="continuationToken">Optional token for pagination to continue listing from a previous operation.</param>
/// <param name="keyPrefix">Optional prefix to filter keys. Only keys starting with this prefix will be returned.</param>
public class ListKeysOperation(string? continuationToken, string? keyPrefix = null) : ActorStateReadOperation
public sealed class ListKeysOperation(string? continuationToken, string? keyPrefix = null) : ActorStateReadOperation
{
/// <summary>
/// Gets the continuation token for pagination.
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/ListKeysResult.cs
+1 -1
View File
@@ -10,7 +10,7 @@ namespace Microsoft.Extensions.AI.Agents.Runtime;
/// </summary>
/// <param name="keys">The collection of keys found in the actor's state.</param>
/// <param name="continuationToken">Optional token for pagination to retrieve additional keys.</param>
public class ListKeysResult(IReadOnlyCollection<string> keys, string? continuationToken) : ActorReadResult
public sealed class ListKeysResult(IReadOnlyCollection<string> keys, string? continuationToken) : ActorReadResult
{
/// <summary>
/// Gets the collection of keys found in the actor's state.
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/MessageContext.cs
-48
View File
@@ -1,48 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Text.Json;
using System.Threading;
using Microsoft.Shared.Diagnostics;
namespace Microsoft.Extensions.AI.Agents.Runtime;
/// <summary>
/// Represents the context of a message being sent within the agent runtime.
/// </summary>
/// <remarks>
/// This includes metadata such as the sender, topic, ahd RPC status.
/// </remarks>
public sealed class MessageContext
{
private string? _messageId;
/// <summary>
/// Gets or sets the unique identifier for this message.
/// </summary>
public string MessageId
{
get => this._messageId ?? Interlocked.CompareExchange(ref this._messageId, Guid.NewGuid().ToString(), null) ?? this._messageId;
set => this._messageId = Throw.IfNullOrEmpty(value);
}
/// <summary>
/// Gets or sets the sender of the message.
/// If <c>null</c>, the sender is unspecified.
/// </summary>
public ActorId? Sender { get; set; }
/// <summary>
/// Gets or sets the topic associated with the message.
/// If <c>null</c>, the message is not tied to a specific topic.
/// </summary>
public TopicId? Topic { get; set; }
/// <summary>
/// Gets or sets a value indicating whether this message is part of an RPC (Remote Procedure Call).
/// </summary>
public bool IsRpc { get; set; }
/// <summary>Gets or sets the serializer options to be used when performing JSON serialization associated with this message.</summary>
public JsonSerializerOptions? SerializerOptions { get; set; }
}
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/ReadResponse.cs
+1 -1
View File
@@ -10,7 +10,7 @@ namespace Microsoft.Extensions.AI.Agents.Runtime;
/// </summary>
/// <param name="eTag">The actor's last-known ETag value.</param>
/// <param name="results">The ordered collection of results.</param>
public class ReadResponse(string eTag, IReadOnlyList<ActorReadResult> results)
public sealed class ReadResponse(string eTag, IReadOnlyList<ActorReadResult> results)
{
/// <summary>
/// Gets the version of the state update.
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/RuntimeActor.cs
-207
View File
@@ -1,207 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Text.Json;
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;
/// <summary>
/// Provides a base implementation of <see cref="IRuntimeActor"/>.
/// </summary>
public abstract class RuntimeActor : IRuntimeActor
{
private static readonly JsonElement s_emptyElement = JsonDocument.Parse("{}").RootElement;
/// <summary>
/// The activity source for tracing.
/// </summary>
public static readonly ActivitySource TraceSource = new($"{typeof(IRuntimeActor).Namespace}");
private readonly Dictionary<Type, HandlerInvoker> _handlerInvokers = [];
private readonly IAgentRuntime _runtime;
private delegate ValueTask<object?> HandlerInvoker(object? message, MessageContext messageContext, CancellationToken cancellationToken);
/// <summary>
/// Provides logging capabilities used for diagnostic and operational information.
/// </summary>
protected internal ILogger Logger { get; }
/// <summary>
/// Gets the unique identifier of the actor.
/// </summary>
public ActorId Id { get; }
/// <summary>
/// Gets the metadata of the actor.
/// </summary>
public ActorMetadata Metadata { get; }
/// <summary>
/// Initializes a new instance of the RuntimeActor class with the specified identifier, runtime, description, and optional logger.
/// </summary>
/// <param name="id">The unique identifier of the actor.</param>
/// <param name="runtime">The runtime environment in which the actor operates.</param>
/// <param name="description">A brief description of the actor's purpose.</param>
/// <param name="logger">An optional logger for recording diagnostic information.</param>
protected RuntimeActor(
ActorId id,
IAgentRuntime runtime,
string? description = null,
ILogger? logger = null)
{
Throw.IfNull(runtime);
this.Id = id;
this._runtime = runtime;
this.Logger = logger ?? NullLogger.Instance;
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>
/// <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>(Func<TInput, MessageContext, CancellationToken, ValueTask> messageHandler)
{
_ = Throw.IfNull(messageHandler);
this.RegisterMessageHandler<TInput, object?>(async (input, ctx, cancellationToken) =>
{
await messageHandler(input, ctx, cancellationToken).ConfigureAwait(false);
return null;
});
}
/// <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>
/// <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, CancellationToken, ValueTask<TOutput>> messageHandler)
{
_ = Throw.IfNull(messageHandler);
if (this._handlerInvokers.ContainsKey(typeof(TInput)))
{
throw new InvalidOperationException($"A handler for type {typeof(TInput)} is already registered.");
}
this._handlerInvokers.Add(
typeof(TInput),
async (message, messageContext, cancellationToken) => await messageHandler((TInput)message!, messageContext, cancellationToken).ConfigureAwait(false));
}
/// <summary>
/// Handles an incoming message by determining its type and invoking the corresponding handler method if available.
/// </summary>
/// <param name="message">The message object to be handled.</param>
/// <param name="messageContext">The context associated with the message.</param>
/// <param name="cancellationToken">A token used to cancel the operation if needed.</param>
/// <returns>A ValueTask that represents the asynchronous operation, containing the response object or null.</returns>
public ValueTask<object?> OnMessageAsync(object message, MessageContext messageContext, CancellationToken cancellationToken = default)
{
// Get the handler for the message type, and invoke it, if it exists.
if (message is not null && this._handlerInvokers.TryGetValue(message.GetType(), out HandlerInvoker? handlerInvoker))
{
return handlerInvoker(message, messageContext, cancellationToken);
}
return new((object?)null);
}
/// <inheritdoc/>
public virtual ValueTask<JsonElement> SaveStateAsync(CancellationToken cancellationToken = default) =>
new(s_emptyElement);
/// <inheritdoc/>
public virtual ValueTask LoadStateAsync(JsonElement state, CancellationToken cancellationToken = default) =>
default;
/// <summary>
/// Sends a message to a specified recipient actor through the runtime.
/// </summary>
/// <param name="actor">The requested actor's type.</param>
/// <param name="cancellationToken">A token used to cancel the operation if needed.</param>
/// <returns>A ValueTask that represents the asynchronous operation, returning the response object or null.</returns>
protected async ValueTask<ActorId?> GetActorAsync(ActorType actor, CancellationToken cancellationToken = default)
{
try
{
return await this._runtime.GetActorAsync(actor, lazy: false, cancellationToken: cancellationToken).ConfigureAwait(false);
}
catch (InvalidOperationException)
{
return null;
}
}
/// <summary>
/// Sends a message to a specified recipient actor through the runtime.
/// </summary>
/// <param name="message">The message object to send.</param>
/// <param name="recipient">The recipient actor's identifier.</param>
/// <param name="messageId">An optional identifier for the message.</param>
/// <param name="cancellationToken">A token used to cancel the operation if needed.</param>
/// <returns>A ValueTask that represents the asynchronous operation, returning the response object or null.</returns>
protected ValueTask<object?> SendMessageAsync(object message, ActorId recipient, string? messageId = null, CancellationToken cancellationToken = default) =>
this._runtime.SendMessageAsync(message, recipient, sender: this.Id, messageId, cancellationToken);
/// <summary>
/// Publishes a message to all actors subscribed to a specific topic through the runtime.
/// </summary>
/// <param name="message">The message object to publish.</param>
/// <param name="topic">The topic identifier to which the message is published.</param>
/// <param name="messageId">An optional identifier for the message.</param>
/// <param name="cancellationToken">A token used to cancel the operation if needed.</param>
/// <returns>A ValueTask that represents the asynchronous publish operation.</returns>
protected ValueTask PublishMessageAsync(object message, TopicId topic, string? messageId = null, CancellationToken cancellationToken = default) =>
this._runtime.PublishMessageAsync(message, topic, sender: this.Id, messageId, cancellationToken);
}
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/TopicId.cs
-110
View File
@@ -1,110 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Diagnostics.CodeAnalysis;
using System.Text.RegularExpressions;
using Microsoft.Shared.Diagnostics;
namespace Microsoft.Extensions.AI.Agents.Runtime;
/// <summary>
/// Provides a topic identifier that defines the scope of a broadcast message.
/// </summary>
/// <remarks>
/// The agent runtime implements a publish-subscribe model through its broadcast API,
/// where messages must be published with a specific topic.
/// </remarks>
public readonly partial struct TopicId : IEquatable<TopicId>
{
private const string TypePattern = @"^[\w-.:=]+$";
#if NET
[GeneratedRegex(TypePattern)]
private static partial Regex TypeRegex();
#else
private static Regex TypeRegex() => s_typeRegex;
private static readonly Regex s_typeRegex = new(TypePattern, RegexOptions.Compiled);
#endif
/// <summary>
/// Initializes a new instance of the <see cref="TopicId"/> struct.
/// </summary>
/// <param name="type">The type of the topic. Must match the pattern: <c>^[\w-.:=]+$</c></param>
/// <param name="source">The source of the event.</param>
public TopicId(string type, string? source = null)
{
Throw.IfNull(type);
if (!TypeRegex().IsMatch(type))
{
Throw.ArgumentException(nameof(type), "Invalid type format.");
}
// TODO: What validation should be performed on source? The cited cloudevents spec suggests it should be a URI reference.
this.Type = type;
this.Source = source ?? "default";
}
/// <summary>
/// Gets the type of the event that this <see cref="TopicId"/> represents.
/// </summary>
/// <remarks>
/// This adheres to the CloudEvents specification.
/// <see href="https://github.com/cloudevents/spec/blob/main/cloudevents/spec.md#type">CloudEvents Type</see>.
/// </remarks>
public string Type { get; }
/// <summary>
/// Gets the source that identifies the context in which an event happened.
/// </summary>
/// <remarks>
/// This adheres to the CloudEvents specification.
/// <see href="https://github.com/cloudevents/spec/blob/main/cloudevents/spec.md#source-1">CloudEvents Source</see>.
/// </remarks>
public string Source { get; }
/// <summary>
/// Convert a string of the format "type/key" into an <see cref="TopicId"/>.
/// </summary>
/// <param name="TopicId">The actor ID string.</param>
/// <returns>An instance of <see cref="TopicId"/>.</returns>
public static TopicId Parse(string TopicId)
{
if (!KeyValueParser.TryParse(TopicId, out string? type, out string? key))
{
throw new FormatException($"Invalid TopicId format: '{TopicId}'. Expected format is 'type/key'.");
}
return new TopicId(type, key);
}
/// <inheritdoc />
public override readonly string ToString() => $"{this.Type}/{this.Source}";
/// <inheritdoc />
public override readonly bool Equals([NotNullWhen(true)] object? obj) =>
obj is TopicId other && this.Equals(other);
/// <inheritdoc/>
public readonly bool Equals(TopicId other) =>
this.Type == other.Type && this.Source == other.Source;
/// <inheritdoc />
public override readonly int GetHashCode() =>
HashCode.Combine(this.Type, this.Source);
/// <inheritdoc />
public static bool operator ==(TopicId left, TopicId right) =>
left.Equals(right);
/// <inheritdoc />
public static bool operator !=(TopicId left, TopicId right) =>
!left.Equals(right);
// TODO: Implement < for wildcard matching (type, *)
//public readonly bool IsWildcardMatch(TopicId other)
//{
// return this.Type == other.Type;
//}
}
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/TypeSubscription.cs
-98
View File
@@ -1,98 +0,0 @@
// Copyright (c) Microsoft. All rights reserved.
using System;
using System.Diagnostics.CodeAnalysis;
using Microsoft.Shared.Diagnostics;
namespace Microsoft.Extensions.AI.Agents.Runtime;
/// <summary>
/// This subscription matches on topics based on the exact type and maps to actors using the source of the topic as the actor key.
/// This subscription causes each source to have its own actor instance.
/// </summary>
/// <remarks>
/// Example:
/// <code>
/// var subscription = new TypeSubscription("t1", "a1");
/// </code>
/// In this case:
/// - A <see cref="TopicId"/> with type `"t1"` and source `"s1"` will be handled by an actor of type `"a1"` with key `"s1"`.
/// - A <see cref="TopicId"/> with type `"t1"` and source `"s2"` will be handled by an actor of type `"a1"` with key `"s2"`.
/// </remarks>
public sealed class TypeSubscription : ISubscriptionDefinition
{
/// <summary>
/// Initializes a new instance of the <see cref="TypeSubscription"/> class.
/// </summary>
/// <param name="topicType">The exact topic type to match against.</param>
/// <param name="actorType">Actor type to handle this subscription.</param>
/// <param name="id">Unique identifier for the subscription. If not provided, a new UUID will be generated.</param>
public TypeSubscription(string topicType, ActorType actorType, string? id = null)
{
Throw.IfNullOrEmpty(topicType);
this.TopicType = topicType;
this.ActorType = actorType;
this.Id = id ?? Guid.NewGuid().ToString();
}
/// <summary>
/// Gets the unique identifier of the subscription.
/// </summary>
public string Id { get; }
/// <summary>
/// Gets the exact topic type used for matching.
/// </summary>
public string TopicType { get; }
/// <summary>
/// Gets the actor type that handles this subscription.
/// </summary>
public ActorType ActorType { get; }
/// <summary>
/// Checks if a given <see cref="TopicId"/> matches the subscription based on an exact type match.
/// </summary>
/// <param name="topic">The topic to check.</param>
/// <returns><c>true</c> if the topic's type matches exactly, <c>false</c> otherwise.</returns>
public bool Matches(TopicId topic) => topic.Type == this.TopicType;
/// <summary>
/// Maps a <see cref="TopicId"/> to an <see cref="ActorId"/>. Should only be called if <see cref="Matches"/> returns true.
/// </summary>
/// <param name="topic">The topic to map.</param>
/// <returns>An <see cref="ActorId"/> representing the actor that should handle the topic.</returns>
/// <exception cref="InvalidOperationException">Thrown if the topic does not match the subscription.</exception>
public ActorId MapToActor(TopicId topic)
{
if (!this.Matches(topic))
{
throw new InvalidOperationException("TopicId does not match the subscription.");
}
return new ActorId(this.ActorType, topic.Source);
}
/// <summary>
/// Determines whether the specified object is equal to the current subscription.
/// </summary>
/// <param name="obj">The object to compare with the current instance.</param>
/// <returns><c>true</c> if the specified object is equal to this instance; otherwise, <c>false</c>.</returns>
public override bool Equals([NotNullWhen(true)] object? obj) =>
obj is TypeSubscription other &&
(this.Id == other.Id || (this.ActorType == other.ActorType && this.TopicType == other.TopicType));
/// <summary>
/// Determines whether the specified subscription is equal to the current subscription.
/// </summary>
/// <param name="other">The subscription to compare.</param>
/// <returns><c>true</c> if the subscriptions are equal; otherwise, <c>false</c>.</returns>
public bool Equals(ISubscriptionDefinition? other) => this.Id == other?.Id;
/// <summary>
/// Returns a hash code for this instance.
/// </summary>
/// <returns>A hash code for this instance, suitable for use in hashing algorithms and data structures.</returns>
public override int GetHashCode() => HashCode.Combine(this.Id, this.ActorType, this.TopicType);
}
dotnet/src/Microsoft.Extensions.AI.Agents.Runtime.Abstractions/WriteResponse.cs
+1 -1
View File
@@ -9,7 +9,7 @@ namespace Microsoft.Extensions.AI.Agents.Runtime;
/// </summary>
/// <param name="eTag">The actor's updated ETag value after the write operation.</param>
/// <param name="success">Whether the write operation was successful.</param>
public class WriteResponse(string eTag, bool success)
public sealed class WriteResponse(string eTag, bool success)
{
/// <summary>
/// Gets the version of the state update.