// 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 }; } }