// Copyright (c) Microsoft. All rights reserved. using System; using System.Collections.Generic; using System.Diagnostics; using System.Threading; using System.Threading.Channels; using System.Threading.Tasks; using Microsoft.Extensions.AI; using Microsoft.Shared.Diagnostics; namespace Microsoft.Agents.AI; /// Represents a delegating AI agent that wraps an inner agent with implementations provided by delegates. /// /// This internal class is a convenience implementation mainly used to support Use methods that take delegates to intercept agent operations. /// internal sealed class AnonymousDelegatingAIAgent : DelegatingAIAgent { /// The delegate to use as the implementation of . private readonly Func, AgentSession?, AgentRunOptions?, AIAgent, CancellationToken, Task>? _runFunc; /// The delegate to use as the implementation of . /// /// When non-, this delegate is used as the implementation of and /// will be invoked with the same arguments as the method itself. /// When , will delegate directly to the inner agent. /// private readonly Func, AgentSession?, AgentRunOptions?, AIAgent, CancellationToken, IAsyncEnumerable>? _runStreamingFunc; /// The delegate to use as the implementation of both and . private readonly Func, AgentSession?, AgentRunOptions?, Func, AgentSession?, AgentRunOptions?, CancellationToken, Task>, CancellationToken, Task>? _sharedFunc; /// /// Initializes a new instance of the class. /// /// The inner agent. /// /// A delegate that provides the implementation for both and . /// In addition to the arguments for the operation, it's provided with a delegate to the inner agent that should be /// used to perform the operation on the inner agent. It will handle both the non-streaming and streaming cases. /// /// /// This overload may be used when the anonymous implementation needs to provide pre-processing and/or post-processing, but doesn't /// need to interact with the results of the operation, which will come from the inner agent. /// /// is . /// is . public AnonymousDelegatingAIAgent( AIAgent innerAgent, Func, AgentSession?, AgentRunOptions?, Func, AgentSession?, AgentRunOptions?, CancellationToken, Task>, CancellationToken, Task> sharedFunc) : base(innerAgent) { _ = Throw.IfNull(sharedFunc); this._sharedFunc = sharedFunc; } /// /// Initializes a new instance of the class. /// /// The inner agent. /// /// A delegate that provides the implementation for . When , /// must be non-null, and the implementation of /// will use for the implementation. /// /// /// A delegate that provides the implementation for . When , /// must be non-null, and the implementation of /// will use for the implementation. /// /// is . /// Both and are . public AnonymousDelegatingAIAgent( AIAgent innerAgent, Func, AgentSession?, AgentRunOptions?, AIAgent, CancellationToken, Task>? runFunc, Func, AgentSession?, AgentRunOptions?, AIAgent, CancellationToken, IAsyncEnumerable>? runStreamingFunc) : base(innerAgent) { ThrowIfBothDelegatesNull(runFunc, runStreamingFunc); this._runFunc = runFunc; this._runStreamingFunc = runStreamingFunc; } /// protected override Task RunCoreAsync( IEnumerable messages, AgentSession? session = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default) { _ = Throw.IfNull(messages); if (this._sharedFunc is not null) { return GetRunViaSharedAsync(messages, session, options, cancellationToken); async Task GetRunViaSharedAsync( IEnumerable messages, AgentSession? session, AgentRunOptions? options, CancellationToken cancellationToken) { AgentResponse? response = null; await this._sharedFunc( messages, session, options, async (messages, session, options, cancellationToken) => response = await this.InnerAgent.RunAsync(messages, session, options, cancellationToken).ConfigureAwait(false), cancellationToken) .ConfigureAwait(false); if (response is null) { Throw.InvalidOperationException("The shared delegate completed successfully without producing an AgentResponse."); } return response; } } else if (this._runFunc is not null) { return this._runFunc(messages, session, options, this.InnerAgent, cancellationToken); } else { Debug.Assert(this._runStreamingFunc is not null, "Expected non-null streaming delegate."); return this._runStreamingFunc!(messages, session, options, this.InnerAgent, cancellationToken) .ToAgentResponseAsync(cancellationToken); } } /// protected override IAsyncEnumerable RunCoreStreamingAsync( IEnumerable messages, AgentSession? session = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default) { _ = Throw.IfNull(messages); if (this._sharedFunc is not null) { var updates = Channel.CreateBounded(1); _ = ProcessAsync(); async Task ProcessAsync() { Exception? error = null; try { await this._sharedFunc(messages, session, options, async (messages, session, options, cancellationToken) => { await foreach (var update in this.InnerAgent.RunStreamingAsync(messages, session, options, cancellationToken).ConfigureAwait(false)) { await updates.Writer.WriteAsync(update, cancellationToken).ConfigureAwait(false); } }, cancellationToken).ConfigureAwait(false); } catch (Exception ex) { error = ex; throw; } finally { _ = updates.Writer.TryComplete(error); } } return updates.Reader.ReadAllAsync(cancellationToken); } else if (this._runStreamingFunc is not null) { return this._runStreamingFunc(messages, session, options, this.InnerAgent, cancellationToken); } else { Debug.Assert(this._runFunc is not null, "Expected non-null non-streaming delegate."); return GetStreamingRunAsyncViaRunAsync(this._runFunc!(messages, session, options, this.InnerAgent, cancellationToken)); static async IAsyncEnumerable GetStreamingRunAsyncViaRunAsync(Task task) { AgentResponse response = await task.ConfigureAwait(false); foreach (var update in response.ToAgentResponseUpdates()) { yield return update; } } } } /// Throws an exception if both of the specified delegates are . /// Both and are . internal static void ThrowIfBothDelegatesNull(object? runFunc, object? runStreamingFunc) { if (runFunc is null && runStreamingFunc is null) { Throw.ArgumentNullException(nameof(runFunc), $"At least one of the {nameof(runFunc)} or {nameof(runStreamingFunc)} delegates must be non-null."); } } }