本教程将演示如何通过实现AIContextProvider并将其附加到代理来添加内存。
重要
并非所有代理类型都支持 AIContextProvider。 此步骤使用的ChatClientAgent确实支持AIContextProvider。
先决条件
有关先决条件和安装 NuGet 包，请参阅本教程中的 “创建并运行简单代理 ”步骤。
创建 AIContextProvider
AIContextProvider是一个抽象类，您可以从中继承，并且可以与AgentThread关联以用于ChatClientAgent。
该功能允许：
- 在代理调用基础推理服务之前和之后运行自定义逻辑。
- 在调用基础推理服务之前，向代理提供其他上下文。
- 检查代理提供和生成的所有消息。
预调用和后调用事件
该 AIContextProvider 类有两种方法，可以重写这些方法以在代理调用基础推理服务之前和之后运行自定义逻辑：
-
InvokingAsync- 在代理调用基础推理服务之前调用。 可以通过返回对象AIContext向代理提供其他上下文。 在调用基础服务之前，此上下文将与代理的现有上下文合并。 可以提供说明、工具和消息以添加到请求中。 -
InvokedAsync- 在代理收到来自基础推理服务的响应后调用。 可以检查请求和响应消息，并更新上下文提供程序的状态。
Serialization
AIContextProvider 实例在创建线程时，以及当线程从序列化状态恢复时被创建并附加到 AgentThread。
实例 AIContextProvider 可能有自己的状态，需要在代理的调用之间保留。 例如，记住有关用户的信息的内存组件可能在其状态中包含记忆。
若要允许线程持久化，需要实现 SerializeAsync 类的 AIContextProvider 方法。 还需要提供采用 JsonElement 参数的构造函数，该构造函数可用于在恢复线程时反序列化状态。
示例 AIContextProvider 实现
下面的自定义内存组件示例会记住用户的名称和年龄，并在每次调用之前将其提供给代理。
首先，创建一个模型类来保存记忆。
internal sealed class UserInfo
{
public string? UserName { get; set; }
public int? UserAge { get; set; }
}
然后，你可以实现 AIContextProvider 来管理内存。
下面的 UserInfoMemory 类包含以下行为：
- 当每次运行结束时将新消息添加到线程时，它使用
IChatClient查找用户消息的名称和年龄。 - 每次调用之前，它都会向代理提供任何当前内存。
- 如果没有可用的内存，它会指示代理询问用户缺少的信息，并在提供信息之前不回答任何问题。
- 它还实现序列化，以允许将内存保留为线程状态的一部分。
using System.Linq;
using System.Text;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;
internal sealed class UserInfoMemory : AIContextProvider
{
private readonly IChatClient _chatClient;
public UserInfoMemory(IChatClient chatClient, UserInfo? userInfo = null)
{
this._chatClient = chatClient;
this.UserInfo = userInfo ?? new UserInfo();
}
public UserInfoMemory(IChatClient chatClient, JsonElement serializedState, JsonSerializerOptions? jsonSerializerOptions = null)
{
this._chatClient = chatClient;
this.UserInfo = serializedState.ValueKind == JsonValueKind.Object ?
serializedState.Deserialize<UserInfo>(jsonSerializerOptions)! :
new UserInfo();
}
public UserInfo UserInfo { get; set; }
public override async ValueTask InvokedAsync(
InvokedContext context,
CancellationToken cancellationToken = default)
{
if ((this.UserInfo.UserName is null || this.UserInfo.UserAge is null) && context.RequestMessages.Any(x => x.Role == ChatRole.User))
{
var result = await this._chatClient.GetResponseAsync<UserInfo>(
context.RequestMessages,
new ChatOptions()
{
Instructions = "Extract the user's name and age from the message if present. If not present return nulls."
},
cancellationToken: cancellationToken);
this.UserInfo.UserName ??= result.Result.UserName;
this.UserInfo.UserAge ??= result.Result.UserAge;
}
}
public override ValueTask<AIContext> InvokingAsync(
InvokingContext context,
CancellationToken cancellationToken = default)
{
StringBuilder instructions = new();
instructions
.AppendLine(
this.UserInfo.UserName is null ?
"Ask the user for their name and politely decline to answer any questions until they provide it." :
$"The user's name is {this.UserInfo.UserName}.")
.AppendLine(
this.UserInfo.UserAge is null ?
"Ask the user for their age and politely decline to answer any questions until they provide it." :
$"The user's age is {this.UserInfo.UserAge}.");
return new ValueTask<AIContext>(new AIContext
{
Instructions = instructions.ToString()
});
}
public override JsonElement Serialize(JsonSerializerOptions? jsonSerializerOptions = null)
{
return JsonSerializer.SerializeToElement(this.UserInfo, jsonSerializerOptions);
}
}
将 AIContextProvider 与代理配合使用
若要使用自定义 AIContextProvider，需要在创建代理时提供一个 AIContextProviderFactory 。 此工厂允许代理为每个线程创建所需的 AIContextProvider 新实例。
创建 ChatClientAgent 时，可以提供一个 ChatClientAgentOptions 对象，该对象允许在提供其他代理选项的基础上额外提供 AIContextProviderFactory。
using System;
using Azure.AI.OpenAI;
using Azure.Identity;
using OpenAI.Chat;
using OpenAI;
ChatClient chatClient = new AzureOpenAIClient(
new Uri("https://<myresource>.openai.azure.com"),
new AzureCliCredential())
.GetChatClient("gpt-4o-mini");
AIAgent agent = chatClient.CreateAIAgent(new ChatClientAgentOptions()
{
Instructions = "You are a friendly assistant. Always address the user by their name.",
AIContextProviderFactory = ctx => new UserInfoMemory(
chatClient.AsIChatClient(),
ctx.SerializedState,
ctx.JsonSerializerOptions)
});
创建新线程时，AIContextProvider 将创建 GetNewThread 并将其附加到该线程。 提取内存后，可以通过线程 GetService 的方法访问内存组件并检查内存。
// Create a new thread for the conversation.
AgentThread thread = agent.GetNewThread();
Console.WriteLine(await agent.RunAsync("Hello, what is the square root of 9?", thread));
Console.WriteLine(await agent.RunAsync("My name is Ruaidhrí", thread));
Console.WriteLine(await agent.RunAsync("I am 20 years old", thread));
// Access the memory component via the thread's GetService method.
var userInfo = thread.GetService<UserInfoMemory>()?.UserInfo;
Console.WriteLine($"MEMORY - User Name: {userInfo?.UserName}");
Console.WriteLine($"MEMORY - User Age: {userInfo?.UserAge}");
本教程演示如何通过实现 ContextProvider 并将其附加到代理来为代理添加内存。
重要
并非所有代理类型都支持 ContextProvider。 此步骤使用的ChatAgent确实支持ContextProvider。
先决条件
有关先决条件和安装包，请参阅本教程中的 “创建并运行简单代理 ”步骤。
创建 ContextProvider
ContextProvider是一个抽象类，您可以从中继承，并且能够与AgentThread关联ChatAgent。
该功能允许：
- 在代理调用基础推理服务之前和之后运行自定义逻辑。
- 在调用基础推理服务之前，向代理提供其他上下文。
- 检查代理提供和生成的所有消息。
预调用和后调用事件
该 ContextProvider 类有两种方法，可以重写这些方法以在代理调用基础推理服务之前和之后运行自定义逻辑：
-
invoking- 在代理调用基础推理服务之前调用。 可以通过返回对象Context向代理提供其他上下文。 在调用基础服务之前，此上下文将与代理的现有上下文合并。 可以提供说明、工具和消息以添加到请求中。 -
invoked- 在代理收到来自基础推理服务的响应后调用。 可以检查请求和响应消息，并更新上下文提供程序的状态。
Serialization
ContextProvider 实例在创建线程时，以及当线程从序列化状态恢复时被创建并附加到 AgentThread。
实例 ContextProvider 可能有自己的状态，需要在代理的调用之间保留。 例如，记住有关用户的信息的内存组件可能在其状态中包含记忆。
若要允许持久化线程，需要为 ContextProvider 类实现序列化。 还需要提供一个构造函数，该构造函数可以在恢复线程时从序列化的数据还原状态。
示例 ContextProvider 实现
下面的自定义内存组件示例会记住用户的名称和年龄，并在每次调用之前将其提供给代理。
首先，创建一个模型类来保存记忆。
from pydantic import BaseModel
class UserInfo(BaseModel):
name: str | None = None
age: int | None = None
然后，你可以实现 ContextProvider 来管理内存。
下面的 UserInfoMemory 类包含以下行为：
- 当每次运行结束时将新消息添加到线程时，它使用聊天客户端查找用户消息的名称和年龄。
- 每次调用之前，它都会向代理提供任何当前内存。
- 如果没有可用的内存，它会指示代理询问用户缺少的信息，并在提供信息之前不回答任何问题。
- 它还实现序列化，以允许将内存保留为线程状态的一部分。
from agent_framework import ContextProvider, Context, InvokedContext, InvokingContext, ChatAgent, ChatClientProtocol
class UserInfoMemory(ContextProvider):
def __init__(self, chat_client: ChatClientProtocol, user_info: UserInfo | None = None, **kwargs: Any):
"""Create the memory.
If you pass in kwargs, they will be attempted to be used to create a UserInfo object.
"""
self._chat_client = chat_client
if user_info:
self.user_info = user_info
elif kwargs:
self.user_info = UserInfo.model_validate(kwargs)
else:
self.user_info = UserInfo()
async def invoked(
self,
request_messages: ChatMessage | Sequence[ChatMessage],
response_messages: ChatMessage | Sequence[ChatMessage] | None = None,
invoke_exception: Exception | None = None,
**kwargs: Any,
) -> None:
"""Extract user information from messages after each agent call."""
# Check if we need to extract user info from user messages
user_messages = [msg for msg in request_messages if hasattr(msg, "role") and msg.role.value == "user"]
if (self.user_info.name is None or self.user_info.age is None) and user_messages:
try:
# Use the chat client to extract structured information
result = await self._chat_client.get_response(
messages=request_messages,
chat_options=ChatOptions(
instructions="Extract the user's name and age from the message if present. If not present return nulls.",
response_format=UserInfo,
),
)
# Update user info with extracted data
if result.value:
if self.user_info.name is None and result.value.name:
self.user_info.name = result.value.name
if self.user_info.age is None and result.value.age:
self.user_info.age = result.value.age
except Exception:
pass # Failed to extract, continue without updating
async def invoking(self, messages: ChatMessage | MutableSequence[ChatMessage], **kwargs: Any) -> Context:
"""Provide user information context before each agent call."""
instructions: list[str] = []
if self.user_info.name is None:
instructions.append(
"Ask the user for their name and politely decline to answer any questions until they provide it."
)
else:
instructions.append(f"The user's name is {self.user_info.name}.")
if self.user_info.age is None:
instructions.append(
"Ask the user for their age and politely decline to answer any questions until they provide it."
)
else:
instructions.append(f"The user's age is {self.user_info.age}.")
# Return context with additional instructions
return Context(instructions=" ".join(instructions))
def serialize(self) -> str:
"""Serialize the user info for thread persistence."""
return self.user_info.model_dump_json()
将 ContextProvider 与代理配合使用
若要使用自定义 ContextProvider，需要在创建代理时提供实例化 ContextProvider 。
创建 ChatAgent 时，可以提供 context_providers 参数以将内存组件附加到代理。
import asyncio
from agent_framework import ChatAgent
from agent_framework.azure import AzureAIAgentClient
from azure.identity.aio import AzureCliCredential
async def main():
async with AzureCliCredential() as credential:
chat_client = AzureAIAgentClient(async_credential=credential)
# Create the memory provider
memory_provider = UserInfoMemory(chat_client)
# Create the agent with memory
async with ChatAgent(
chat_client=chat_client,
instructions="You are a friendly assistant. Always address the user by their name.",
context_providers=memory_provider,
) as agent:
# Create a new thread for the conversation
thread = agent.get_new_thread()
print(await agent.run("Hello, what is the square root of 9?", thread=thread))
print(await agent.run("My name is Ruaidhrí", thread=thread))
print(await agent.run("I am 20 years old", thread=thread))
# Access the memory component via the thread's context_providers attribute and inspect the memories
user_info_memory = thread.context_provider.providers[0]
if user_info_memory:
print()
print(f"MEMORY - User Name: {user_info_memory.user_info.name}")
print(f"MEMORY - User Age: {user_info_memory.user_info.age}")
if __name__ == "__main__":
asyncio.run(main())