// Copyright (c) Microsoft. All rights reserved.
using System.Runtime.CompilerServices;
using System.Text.Json;
using Microsoft.DurableTask;
using Microsoft.DurableTask.Entities;
using Microsoft.Extensions.AI;
using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.AI.DurableTask;
///
/// A durable AIAgent implementation that uses entity methods to interact with agent entities.
///
public sealed class DurableAIAgent : AIAgent
{
private readonly TaskOrchestrationContext _context;
private readonly string _agentName;
///
/// Initializes a new instance of the class.
///
/// The orchestration context.
/// The name of the agent.
internal DurableAIAgent(TaskOrchestrationContext context, string agentName)
{
this._context = context;
this._agentName = agentName;
}
///
/// Creates a new agent session for this agent using a random session ID.
///
/// The cancellation token.
/// A value task that represents the asynchronous operation. The task result contains a new agent session.
protected override ValueTask CreateSessionCoreAsync(CancellationToken cancellationToken = default)
{
AgentSessionId sessionId = this._context.NewAgentSessionId(this._agentName);
return ValueTask.FromResult(new DurableAgentSession(sessionId));
}
///
/// Serializes an agent session to JSON.
///
/// The session to serialize.
/// Optional JSON serializer options.
/// The cancellation token.
/// A containing the serialized session state.
protected override ValueTask SerializeSessionCoreAsync(AgentSession session, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default)
{
if (session is null)
{
throw new ArgumentNullException(nameof(session));
}
if (session is not DurableAgentSession durableSession)
{
throw new InvalidOperationException($"The provided session type '{session.GetType().Name}' is not compatible with this agent. Only sessions of type '{nameof(DurableAgentSession)}' can be serialized by this agent.");
}
return new(durableSession.Serialize(jsonSerializerOptions));
}
///
/// Deserializes an agent session from JSON.
///
/// The serialized session data.
/// Optional JSON serializer options.
/// The cancellation token.
/// A value task that represents the asynchronous operation. The task result contains the deserialized agent session.
protected override ValueTask DeserializeSessionCoreAsync(
JsonElement serializedState,
JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default)
{
return ValueTask.FromResult(DurableAgentSession.Deserialize(serializedState, jsonSerializerOptions));
}
///
/// Runs the agent with messages and returns the response.
///
/// The messages to send to the agent.
/// The agent session to use.
/// Optional run options.
/// The cancellation token.
/// The response from the agent.
/// Thrown when the agent has not been registered.
/// Thrown when the provided session is not valid for a durable agent.
/// Thrown when cancellation is requested (cancellation is not supported for durable agents).
protected override async Task RunCoreAsync(
IEnumerable messages,
AgentSession? session = null,
AgentRunOptions? options = null,
CancellationToken cancellationToken = default)
{
if (cancellationToken != default && cancellationToken.CanBeCanceled)
{
throw new NotSupportedException("Cancellation is not supported for durable agents.");
}
session ??= await this.CreateSessionAsync(cancellationToken).ConfigureAwait(false);
if (session is not DurableAgentSession durableSession)
{
throw new ArgumentException(
"The provided session is not valid for a durable agent. " +
"Create a new session using CreateSessionAsync or provide a session previously created by this agent.",
paramName: nameof(session));
}
IList? enableToolNames = null;
bool enableToolCalls = true;
ChatResponseFormat? responseFormat = null;
if (options is DurableAgentRunOptions durableOptions)
{
enableToolCalls = durableOptions.EnableToolCalls;
enableToolNames = durableOptions.EnableToolNames;
}
else if (options is ChatClientAgentRunOptions chatClientOptions && chatClientOptions.ChatOptions?.Tools != null)
{
// Honor the response format from the chat client options if specified
responseFormat = chatClientOptions.ChatOptions?.ResponseFormat;
}
// Override the response format if specified in the agent run options
if (options?.ResponseFormat is { } format)
{
responseFormat = format;
}
RunRequest request = new([.. messages], responseFormat, enableToolCalls, enableToolNames)
{
OrchestrationId = this._context.InstanceId
};
try
{
return await this._context.Entities.CallEntityAsync(
durableSession.SessionId,
nameof(AgentEntity.Run),
request);
}
catch (EntityOperationFailedException e) when (e.FailureDetails.ErrorType == "EntityTaskNotFound")
{
throw new AgentNotRegisteredException(this._agentName, e);
}
}
///
/// Runs the agent with messages and returns a simulated streaming response.
///
///
/// Streaming is not supported for durable agents, so this method just returns the full response
/// as a single update.
///
/// The messages to send to the agent.
/// The agent session to use.
/// Optional run options.
/// The cancellation token.
/// A streaming response enumerable.
protected override async IAsyncEnumerable RunCoreStreamingAsync(
IEnumerable messages,
AgentSession? session = null,
AgentRunOptions? options = null,
[EnumeratorCancellation] CancellationToken cancellationToken = default)
{
// Streaming is not supported for durable agents, so we just return the full response
// as a single update.
AgentResponse response = await this.RunAsync(messages, session, options, cancellationToken);
foreach (AgentResponseUpdate update in response.ToAgentResponseUpdates())
{
yield return update;
}
}
///
/// Run the agent with no message assuming that all required instructions are already provided to the agent or on the session, and requesting a response of the specified type .
///
/// The type of structured output to request.
///
/// The conversation session to use for this invocation. If , a new session will be created.
/// The session will be updated with any response messages generated during invocation.
///
/// Optional JSON serializer options to use for deserializing the response.
/// Optional configuration parameters for controlling the agent's invocation behavior.
/// The to monitor for cancellation requests. The default is .
/// A task that represents the asynchronous operation. The task result contains an with the agent's output.
///
/// This method is specific to durable agents because the Durable Task Framework uses a custom
/// synchronization context for orchestration execution, and all continuations must run on the
/// orchestration thread to avoid breaking the durable orchestration and potential deadlocks.
///
public new Task> RunAsync(
AgentSession? session = null,
JsonSerializerOptions? serializerOptions = null,
AgentRunOptions? options = null,
CancellationToken cancellationToken = default) =>
this.RunAsync([], session, serializerOptions, options, cancellationToken);
///
/// Runs the agent with a text message from the user, requesting a response of the specified type .
///
/// The type of structured output to request.
/// The user message to send to the agent.
///
/// The conversation session to use for this invocation. If , a new session will be created.
/// The session will be updated with the input message and any response messages generated during invocation.
///
/// Optional JSON serializer options to use for deserializing the response.
/// Optional configuration parameters for controlling the agent's invocation behavior.
/// The to monitor for cancellation requests. The default is .
/// A task that represents the asynchronous operation. The task result contains an with the agent's output.
/// is , empty, or contains only whitespace.
///
///
///
public new Task> RunAsync(
string message,
AgentSession? session = null,
JsonSerializerOptions? serializerOptions = null,
AgentRunOptions? options = null,
CancellationToken cancellationToken = default)
{
_ = Throw.IfNull(message);
return this.RunAsync(new ChatMessage(ChatRole.User, message), session, serializerOptions, options, cancellationToken);
}
///
/// Runs the agent with a single chat message, requesting a response of the specified type .
///
/// The type of structured output to request.
/// The chat message to send to the agent.
///
/// The conversation session to use for this invocation. If , a new session will be created.
/// The session will be updated with the input message and any response messages generated during invocation.
///
/// Optional JSON serializer options to use for deserializing the response.
/// Optional configuration parameters for controlling the agent's invocation behavior.
/// The to monitor for cancellation requests. The default is .
/// A task that represents the asynchronous operation. The task result contains an with the agent's output.
/// is .
///
///
///
public new Task> RunAsync(
ChatMessage message,
AgentSession? session = null,
JsonSerializerOptions? serializerOptions = null,
AgentRunOptions? options = null,
CancellationToken cancellationToken = default)
{
_ = Throw.IfNull(message);
return this.RunAsync([message], session, serializerOptions, options, cancellationToken);
}
///
/// Runs the agent with a collection of chat messages, requesting a response of the specified type .
///
/// The type of structured output to request.
/// The collection of messages to send to the agent for processing.
///
/// The conversation session to use for this invocation. If , a new session will be created.
/// The session will be updated with the input messages and any response messages generated during invocation.
///
/// Optional JSON serializer options to use for deserializing the response.
/// Optional configuration parameters for controlling the agent's invocation behavior.
/// The to monitor for cancellation requests. The default is .
/// A task that represents the asynchronous operation. The task result contains an with the agent's output.
///
///
///
public new async Task> RunAsync(
IEnumerable messages,
AgentSession? session = null,
JsonSerializerOptions? serializerOptions = null,
AgentRunOptions? options = null,
CancellationToken cancellationToken = default)
{
serializerOptions ??= AgentAbstractionsJsonUtilities.DefaultOptions;
var responseFormat = ChatResponseFormat.ForJsonSchema(serializerOptions);
(responseFormat, bool isWrappedInObject) = StructuredOutputSchemaUtilities.WrapNonObjectSchema(responseFormat);
options = options?.Clone() ?? new DurableAgentRunOptions();
options.ResponseFormat = responseFormat;
// ConfigureAwait(false) cannot be used here because the Durable Task Framework uses
// a custom synchronization context that requires all continuations to execute on the
// orchestration thread. Scheduling the continuation on an arbitrary thread would break
// the orchestration.
AgentResponse response = await this.RunAsync(messages, session, options, cancellationToken);
return new AgentResponse(response, serializerOptions) { IsWrappedInObject = isWrappedInObject };
}
}