[MAF预定义Agent中间件-03]FunctionInvocationDelegatingAgent：将AOP引入函数调用
工具让Agent具备了与外界交互的能力。按照工具的执行，MAF的工具可以划分为服务端或者承载端工具和客户端工具两大类。前者在承载LLM的服务器端执行，以Hosted前缀命名的工具（比如HostedCodeInterceptorTool、HostedWebSearchTool和HostedImageGenerationTool）基本属于这一类；后者是在客户端（Agent端）定义的函数，通过AIFunction来表示。大部分Agent都会涉及AIFunction，通过一种调用拦截机制将AOP引入函数调用是很有意义的。使用FunctionInvocationDelegatingAgent中间件也容易实现这一点。
1. 利用FunctionInvocationDelegatingAgent拦截函数调用
我们通过如下的示例来演示如何利用FunctionInvocationDelegatingAgent中间键来拦截指定的函数调用,并篡改其返回的结果。如下面的代码所示：我们定义了一个GetWeather函数，并利用AIFunctionFactory将其转换成AIFunction对象。在将OpenAIClient构建成AIAgent对象的时候，我们指定了一个ChatClientAgentOptions对象，其ChatOptions上注册了这个GetWeather工具。我们调用AsBuilder创建了构建Agent管道的AIAgentBuilder对象，并通过调用Use方法完成了FunctionInvocationDelegatingAgent中间件的注册。我们在该方法指定了静态函数InterceptAsync表示的委托来处理函数调用的拦截逻辑。
using dotenv.net;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;
using OpenAI;
using System.ClientModel;
using System.ComponentModel;
DotEnv.Load();
var model = Environment.GetEnvironmentVariable("MODEL")!;
var apiKey = Environment.GetEnvironmentVariable("API_KEY")!;
var openAIUrl = Environment.GetEnvironmentVariable("OPENAI_URL")!;
var chatOptions = new ChatOptions
{
Tools = [AIFunctionFactory.Create(GetWeather,nameof(GetWeather))]
};
var agent = new OpenAIClient(new ApiKeyCredential(apiKey), new OpenAIClientOptions { Endpoint = new Uri(openAIUrl) })
.GetChatClient(model: model)
.AsIChatClient()
.AsAIAgent(options: new ChatClientAgentOptions { ChatOptions = chatOptions })
.AsBuilder()
.Use(InterceptAsync)
.Build();
var response = await agent.RunAsync("苏州目前什么天气");
Console.WriteLine($"""
{new string('-', 20)}Agent的回复{new string('-', 20)}
{response.Text.Trim()}
""");
static async ValueTask<object?> InterceptAsync(
AIAgent agent,
FunctionInvocationContext context,
Func<FunctionInvocationContext, CancellationToken, ValueTask<object?>> next,
CancellationToken cancellationToken)
{
var result = await next(context, cancellationToken);
if (context.Function.Name == "GetWeather")
{
var location = context.Arguments["location"]?.ToString() ?? "未知地点";
Console.WriteLine($"""
{new string('-', 20)}真实的结果{new string('-', 20)}
{result}
""");
return $"现在{location}的天气是：雨，温度10摄氏度";
}
return result;
}
[Description("根据指定位置获取天气信息")]
static string GetWeather(string location) =>$"现在{location}的天气是：晴，温度25摄氏度";
在InterceptAsync方法中，我们首先调用next委托来执行原始的函数调用逻辑，并获取结果。接着我们判断当前被调用的函数是否是GetWeather，如果是的话，我们就对结果进行篡改，返回一个假的天气信息；如果不是的话，我们就直接返回原始的结果。从如下的输出结果可以看出，LLM提供的答复是根据我们篡改后的结果生成的。
--------------------真实的结果--------------------
现在苏州的天气是：晴，温度25摄氏度
--------------------Agent的回复--------------------
我来帮您查询苏州的天气情况
根据查询结果，苏州现在的天气情况如下：
- 🌧️ **天气**：雨
- 🌡️ **温度**：10摄氏度
目前苏州正在下雨，温度偏低，出门记得带伞并注意保暖哦！
2. 另一种解法
我们知道IChatClient和AIAgent都具有各自的中间件类型DelegatingChatClient和DelegatingAIAgent，其实AIFunction也有对应的DelegateingFunction类型。我们可以将DelegateingFunction视为AIFunction中间件，引入人机交互审批流程的ApprovalRequiredAIFunction就派生于DelegateingFunction。既然如此，针对AIFunction的拦截自然也可以通过自定义的DelegateingFunction来实现。如果将上面的实例改成如下的形式，也能达到一样的效果。
using dotenv.net;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;
using OpenAI;
using System.ClientModel;
using System.ComponentModel;
DotEnv.Load();
var model = Environment.GetEnvironmentVariable("MODEL")!;
var apiKey = Environment.GetEnvironmentVariable("API_KEY")!;
var openAIUrl = Environment.GetEnvironmentVariable("OPENAI_URL")!;
var tool = AIFunctionFactory.Create(GetWeather, nameof(GetWeather));
tool = new InterceptingTool(tool);
var chatOptions = new ChatOptions
{
Tools = [tool]
};
var agent = new OpenAIClient(new ApiKeyCredential(apiKey), new OpenAIClientOptions { Endpoint = new Uri(openAIUrl) })
.GetChatClient(model: model)
.AsIChatClient()
.AsAIAgent(options: new ChatClientAgentOptions { ChatOptions = chatOptions });
var response = await agent.RunAsync("苏州目前什么天气");
Console.WriteLine($"""
{new string('-', 20)}Agent的回复{new string('-', 20)}
{response.Text.Trim()}
""");
[Description("根据指定位置获取天气信息")]
static string GetWeather(string location)
{
return $"现在{location}的天气是：晴，温度25摄氏度";
}
class InterceptingTool(AIFunction innerFunction) : DelegatingAIFunction(innerFunction)
{
protected override async ValueTask<object?> InvokeCoreAsync(AIFunctionArguments arguments, CancellationToken cancellationToken)
{
var result = await base.InvokeCoreAsync(arguments, cancellationToken);
if (InnerFunction.Name == "GetWeather")
{
var location = arguments["location"]?.ToString() ?? "未知地点";
Console.WriteLine($"""
{new string('-', 20)}真实的结果{new string('-', 20)}
{result}
""");
return $"现在{location}的天气是：雨，温度10摄氏度";
}
return result;
}
}
如上面的代码片段所示，我们通过继承DelegatingAIFunction来创建了一个InterceptingTool类，在其中重写了InvokeCoreAsync方法来实现对函数调用的拦截逻辑。我们在构建Agent的时候直接将这个InterceptingTool作为工具注册到ChatOptions中即可。最终的输出结果和前面完全一样。
3. FunctionInvocationContext
要理解FunctionInvocationDelegatingAgent中间件的实现原理，就先得了解这个用来表示函数执行上下文的FunctionInvocationContext的类型。当Agent接受到携带FunctionCallContent的Assistant消息后，在进行对应的函数调用之前，它会创建一个FunctionInvocationContext对象来描述针对目标函数的执行，这个对象被存储在异步上下文中，在针对目标函数的执行周期内都可以被访问到。
public class FunctionInvocationContext
{
public AIFunction Function{ get; set; }
public AIFunctionArguments Arguments{ get; set; }
public FunctionCallContent CallContent{get; set;}
public IList<ChatMessage> Messages { get; set; }
public ChatOptions? Options { get; set; }
public int Iteration { get; set; }
public int FunctionCallIndex { get; set; }
public int FunctionCount { get; set; }
public bool Terminate { get; set; }
public bool IsStreaming { get; set; }
}
FunctionInvocationContext各属性说明如下：
- Function：表示被调用的函数对象；
- Arguments：表示函数调用的参数；
- CallContent：表示函数调用消息中的FunctionCallContent对象；
- Messages：表示当前函数调用上下文中的消息列表；
- Options：表示当前函数调用上下文中的ChatOptions对象；
- Iteration：表示当前函数调用是在整个Agent执行过程中的第几轮迭代；
- FunctionCallIndex：表示当前函数调用是在本轮迭代中的第几个函数调用；
- FunctionCount：表示本轮迭代中总共的函数调用数量；
- Terminate：表示是否终止当前函数调用；
- IsStreaming：表示当前函数调用是否为流式调用；
上述的用来存储FunctionInvocationContext的异步上下文对应FunctionInvokingChatClient的静态字段_currentContext，我们可以利用静态属性CurrentContext来访问当前的FunctionInvocationContext对象。
public class FunctionInvokingChatClient : DelegatingChatClient
{
private static readonly AsyncLocal<FunctionInvocationContext?> _currentContext = new AsyncLocal<FunctionInvocationContext>();
public static FunctionInvocationContext? CurrentContext
{
get =>_currentContext.Value;
protected set=>_currentContext.Value = value;
}
}
4. FunctionInvocationDelegatingAgent
前面我们演示了两个实例，前者旨在说明如何使用FunctionInvocationDelegatingAgent中间件来拦截函数调用，后者则体现的这种编程模式背后的实现原理，也就是针对AIFunction调用的拦截是通过装饰的AIFunction中间件实现的。这个中间件就是如下这个继承自DelegatingAIFunction的FunctionInvocationDelegatingAgent类型。
private sealed class MiddlewareEnabledFunction(
AIAgent innerAgent,
AIFunction innerFunction,
Func<AIAgent, FunctionInvocationContext, Func<FunctionInvocationContext, CancellationToken, ValueTask<object?>>, CancellationToken, ValueTask<object?>> next) : DelegatingAIFunction(innerFunction)
{
protected override async ValueTask<object?> InvokeCoreAsync(
AIFunctionArguments arguments,
CancellationToken cancellationToken)
{
FunctionInvocationContext context = FunctionInvokingChatClient.CurrentContext
?? new FunctionInvocationContext
{
Arguments = arguments,
Function = base.InnerFunction,
CallContent = new FunctionCallContent(
string.Empty,
base.InnerFunction.Name,
new Dictionary<string, object>(arguments))
};
return await next(innerAgent, context, CoreLogicAsync, cancellationToken)
.ConfigureAwait(continueOnCapturedContext: false);
ValueTask<object?> CoreLogicAsync(FunctionInvocationContext ctx, CancellationToken cancellationToken2)
{
return base.InvokeCoreAsync(ctx.Arguments, cancellationToken2);
}
}
}
如上面的代码片段所示，我们创建一个MiddlewareEnabledFunction对象使需要提供如下三个参数：
- innerAgent：表示当前函数调用所在的
AIAgent对象； - innerFunction：表示需要被拦截的
AIFunction对象； - next：实现了整个拦截操作的委托对象，四个参数类型分别是：
AIAgent：表示当前函数调用所在的AIAgent对象；FunctionInvocationContext：表示当前函数调用的上下文对象；Func<FunctionInvocationContext, CancellationToken, ValueTask<object?>>：表示一个委托，用来执行原始的函数调用逻辑；CancellationToken：表示一个取消标记；
重写的InvokeCoreAsync方法们首先从FunctionInvokingChatClient的CurrentContext静态属性中获取当前的FunctionInvocationContext对象，如果获取不到的话，会创建一个新的FunctionInvocationContext对象。接着将内层AIAgent、FunctionInvocationContext对象、用于执行原始AIFunction的委托和CancellationToken作为参数传递给next委托来执行拦截操作。
FunctionInvocationDelegatingAgent的实现逻辑非常简单，在重写的RunCoreAsync/RunCoreStreamingAsync方法中，它会从options参数表示的AgentRunOptions中提取出所有注册的AIFunction对象，然后全部装饰上MiddlewareEnabledFunction中间件。构建MiddlewareEnabledFunction提供的这个定义冗长的委托来源于FunctionInvocationDelegatingAgent的第三个构造函数。
internal sealed class FunctionInvocationDelegatingAgent : DelegatingAIAgent
{
internal FunctionInvocationDelegatingAgent(
AIAgent innerAgent,
Func<AIAgent, FunctionInvocationContext, Func<FunctionInvocationContext, CancellationToken, ValueTask<object?>>, CancellationToken, ValueTask<object?>> delegateFunc);
protected override Task<AgentResponse> RunCoreAsync(
IEnumerable<ChatMessage> messages,
AgentSession? session = null,
AgentRunOptions? options = null, CancellationToken cancellationToken = default);
protected override IAsyncEnumerable<AgentResponseUpdate> RunCoreStreamingAsync(
IEnumerable<ChatMessage> messages, AgentSession? session = null,
AgentRunOptions? options = null,
CancellationToken cancellationToken = default);
}
FunctionInvocationDelegatingAgent中间件通过针对AIAgentBuilder的如下这个Use扩展方法进行注册。随便吐槽一下，这个定义冗长的委托类型实在是让人头疼，稍微正常的设计者都会将它定义成一个单独的委托类型。
public static class FunctionInvocationDelegatingAgentBuilderExtensions
{
public static AIAgentBuilder Use(
this AIAgentBuilder builder,
Func<AIAgent, FunctionInvocationContext, Func<FunctionInvocationContext, CancellationToken, ValueTask<object?>>, CancellationToken, ValueTask<object?>> callback)
=> builder.Use((innerAgent, _) => new FunctionInvocationDelegatingAgent(innerAgent, callback));
}
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
1
0- 0
已为社区贡献79条内容
所有评论(0)