// 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;
///
/// Provides a base implementation of .
///
public abstract class RuntimeActor : IRuntimeActor
{
private static readonly JsonElement s_emptyElement = JsonDocument.Parse("{}").RootElement;
///
/// The activity source for tracing.
///
public static readonly ActivitySource TraceSource = new($"{typeof(IRuntimeActor).Namespace}");
private readonly Dictionary _handlerInvokers = [];
private readonly IAgentRuntime _runtime;
private delegate ValueTask HandlerInvoker(object? message, MessageContext messageContext, CancellationToken cancellationToken);
///
/// Provides logging capabilities used for diagnostic and operational information.
///
protected internal ILogger Logger { get; }
///
/// Gets the unique identifier of the actor.
///
public ActorId Id { get; }
///
/// Gets the metadata of the actor.
///
public ActorMetadata Metadata { get; }
///
/// Initializes a new instance of the RuntimeActor class with the specified identifier, runtime, description, and optional logger.
///
/// The unique identifier of the actor.
/// The runtime environment in which the actor operates.
/// A brief description of the actor's purpose.
/// An optional logger for recording diagnostic information.
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);
}
/// Registers a handler for .
/// The type of the input message for the handler.
/// The handler function that processes the message.
/// Thrown when a handler for the specified type is already registered.
///
/// The base implementation of will use these registered handlers to process incoming messages.
///
protected void RegisterMessageHandler(Action messageHandler)
{
_ = Throw.IfNull(messageHandler);
this.RegisterMessageHandler(async (input, ctx, cancellationToken) =>
{
messageHandler(input, ctx);
return null;
});
}
/// Registers a handler for .
/// The type of the input message for the handler.
/// The handler function that processes the message.
/// Thrown when a handler for the specified type is already registered.
///
/// The base implementation of will use these registered handlers to process incoming messages.
///
protected void RegisterMessageHandler(Func messageHandler)
{
_ = Throw.IfNull(messageHandler);
this.RegisterMessageHandler(async (input, ctx, cancellationToken) =>
{
await messageHandler(input, ctx, cancellationToken).ConfigureAwait(false);
return null;
});
}
/// Registers a handler for .
/// The type of the input message for the handler.
/// The type of the output message for the handler.
/// The handler function that processes the message.
/// Thrown when a handler for the specified type is already registered.
///
/// The base implementation of will use these registered handlers to process incoming messages.
///
protected void RegisterMessageHandler(Func messageHandler)
{
_ = Throw.IfNull(messageHandler);
this.RegisterMessageHandler(async (input, ctx, cancellationToken) => messageHandler(input, ctx));
}
/// Registers a handler for that produces a .
/// The type of the input message for the handler.
/// The type of the output message for the handler.
/// The handler function that processes the message.
/// Thrown when a handler for the specified type is already registered.
///
/// The base implementation of will use these registered handlers to process incoming messages.
///
protected void RegisterMessageHandler(Func> 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));
}
///
/// Handles an incoming message by determining its type and invoking the corresponding handler method if available.
///
/// The message object to be handled.
/// The context associated with the message.
/// A token used to cancel the operation if needed.
/// A ValueTask that represents the asynchronous operation, containing the response object or null.
public ValueTask 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);
}
///
public virtual ValueTask SaveStateAsync(CancellationToken cancellationToken = default) =>
new(s_emptyElement);
///
public virtual ValueTask LoadStateAsync(JsonElement state, CancellationToken cancellationToken = default) =>
default;
///
/// Sends a message to a specified recipient actor through the runtime.
///
/// The requested actor's type.
/// A token used to cancel the operation if needed.
/// A ValueTask that represents the asynchronous operation, returning the response object or null.
protected async ValueTask GetActorAsync(ActorType actor, CancellationToken cancellationToken = default)
{
try
{
return await this._runtime.GetActorAsync(actor, lazy: false, cancellationToken: cancellationToken).ConfigureAwait(false);
}
catch (InvalidOperationException)
{
return null;
}
}
///
/// Sends a message to a specified recipient actor through the runtime.
///
/// The message object to send.
/// The recipient actor's identifier.
/// An optional identifier for the message.
/// A token used to cancel the operation if needed.
/// A ValueTask that represents the asynchronous operation, returning the response object or null.
protected ValueTask SendMessageAsync(object message, ActorId recipient, string? messageId = null, CancellationToken cancellationToken = default) =>
this._runtime.SendMessageAsync(message, recipient, sender: this.Id, messageId, cancellationToken);
///
/// Publishes a message to all actors subscribed to a specific topic through the runtime.
///
/// The message object to publish.
/// The topic identifier to which the message is published.
/// An optional identifier for the message.
/// A token used to cancel the operation if needed.
/// A ValueTask that represents the asynchronous publish operation.
protected ValueTask PublishMessageAsync(object message, TopicId topic, string? messageId = null, CancellationToken cancellationToken = default) =>
this._runtime.PublishMessageAsync(message, topic, sender: this.Id, messageId, cancellationToken);
}