// 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.");
}
}
}